spark1.6+hadoop2.6+kafka2.10-0.8.2.1+zookeeper3.3.6安裝及sparkStreaming程式碼編寫和除錯
阿新 • • 發佈：2019-02-02
安裝環境
安裝之前確保裝置至少有4GB記憶體,推薦8GB
centos7.2
docker(這個安裝請參考我的另一篇部落格https://blog.csdn.net/qq_16563637/article/details/81699251)
| 目標安裝軟體 | 目標安裝版本 | 實際安裝版本 |
|---|---|---|
| hadoop | 2.6 | 2.6 |
| spark | 1.6 | 1.6 |
| kafka | 2.10-0.8.2.1 | 2.10-0.8.2.1 |
| zookeeper | 3.3.3 | 3.3.6 |
說明 因為kafka要配置HOST_NAME,HOST_NAME必須是宿主機IP地址,否則消費者程式會根據容器IP地址尋找kafka,所以kafka不能使用docker安裝
本人親自測試能夠正常使用
-------------------------------------------開始安裝spark和hadoop------------------------------------------------------
下載spark(說明這個說明:spark安裝時會自動安裝hadoop,不用再單獨安裝hadoop)
docker pull registry.docker-cn.com/sequenceiq/spark:1.6.0
docker執行
docker run -it -p 4040:4040 -p 7077:7077 -p 8088:8088 -p 8081:8081 -p 8080:8080 -p 8042:8042 -p 8030:8030 -p 8031:8031 -p 8040:8040 -p 9000:9000 -p 49707:49707 -p 50010:50010 -p 50070:50070 -p 50075:50075 -p 50020:50020 -p 50090:50090 --name spark --rm sequenceiq/spark:1.6.0 /bin/bash
設定spark
進入容器(docker run執行後會直接進入容器,該步驟可以省略)
docker exec -it 容器名 /bin/bash
docker exec -it 05d499dd260f /bin/bash
cd /usr/local/spark-1.6.0-bin-hadoop2.6
cd conf
cp spark-env.sh.template spark-env.sh
vi spark-env.sh
在最底部新增
export JAVA_HOME=/usr/java/jdk1.7.0_51
export SPARK_MASTER_PORT=7077
儲存
cp slaves.template slaves vi slaves
去掉localhost
新增192.168.1.103
儲存
cd ../sbin
./stop-all.sh
./start-master.sh
./start-slave.sh 192.168.1.103:7077 --webui-port 8081
為了正常使用配置環境變數,進入容器中設定
docker exec -it 05d499dd260f /bin/bash
vi /etc/profile
export SPARK_HOME=/usr/local/spark-1.6.0-bin-hadoop2.6
export HADOOP_HOME="/usr/local/hadoop-2.6.0"
export PATH=$PATH:$SPARK_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
儲存:wq
使配置檔案生效
source /etc/profile
測試spark是否安裝OK
執行下面命令(該演算法是利用蒙特·卡羅演算法求PI)
/usr/local/spark-1.6.0-bin-hadoop2.6/bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://192.168.1.103:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
/usr/local/spark-1.6.0-bin-hadoop2.6/lib/spark-examples-1.6.0-hadoop2.6.0.jar \
100
如果沒有報錯說明spark安裝正常
開啟spark-shell
spark-shell
出現 scala> 說明安裝正常
開始檢查hadoop是否正常
ctrl+C退出spark-shell
檢視命令列是否能用
hadoop version
如果輸出版本資訊繼續輸入
cd /usr/local/hadoop-2.6.0
bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.6.0.jar grep input output 'dfs[a-z.]+'
如果輸出mapreduce程式執行說明安裝正常
檢視spark 管控臺
http://192.168.1.103:8080/
檢視hadoop 管控臺
http://192.168.1.103:50070
結束
------------------------------------------------zookeeper的docker安裝------------------------------------------------------
說明:kafka_2.10-0.8.2.1推薦的zookeeper版本為3.3.3,然而zookeeper官方映象沒有3.3.3,選用3.3.6安裝
docker pull zookeeper:3.3.6
啟動zookeeper容器
docker run -d --name zookeeper -p 2181:2181 -t zookeeper:3.3.6
zookeeper安裝完成
------------------------------------------------kafka本地安裝--------------------------------------------------------------
先安裝jdk(可參考我的另一篇部落格https://blog.csdn.net/qq_16563637/article/details/81738113)
安裝成功後下載kafka_2.10-0.8.2.1.tgz
kafka下載地址:https://archive.apache.org/dist/kafka/0.8.2.1/kafka_2.10-0.8.2.1.tgz
將檔案上傳至伺服器解壓縮
tar zxf kafka_2.10-0.8.2.1.tgz
修改配置檔案
cd /root/kafka_2.10-0.8.2.1/config
vi server.properties
修改下面幾項內容
host.name=192.168.1.103
log.dirs=/root/kafka_2.10-0.8.2.1/logs
zookeeper.connect=192.168.1.103:2181
儲存
cd ..
啟動kafka(前臺啟動)
建議先前臺啟動觀看日誌沒有報錯,ctrl+c退出,再後臺啟動
bin/kafka-server-start.sh config/server.properties
啟動kafka(後臺啟動)
bin/kafka-server-start.sh config/server.properties > /dev/null 2>&1 &
建立topic(此處partitions數量不能大於broker,replication-factor 為副本數量)
bin/kafka-topics.sh --create --zookeeper 192.168.1.103:2181 --replication-factor 1 --partitions 1 --topic test
列出所有topic
bin/kafka-topics.sh --list --zookeeper 192.168.1.103:2181
向topic中寫入資料
bin/kafka-console-producer.sh --broker-list 192.168.1.103:9092 --topic test
消費資料
bin/kafka-console-consumer.sh --zookeeper 192.168.1.103:2181 --topic test --from-beginning
檢視指定topic的詳情
bin/kafka-topics.sh --describe --zookeeper 192.168.1.103:2181 --topic test
停止kafka
kill -s TERM $(jps -l | grep 'kafka\.Kafka' | awk '{print $1}')
----------------------------------------------spark測試程式碼---------------------------------------------------------
說明:下面這段程式碼採用maven專案,請直接在本地啟動執行,並且在Program arguments中寫入下面內容
192.168.1.103:2181 g1 test 2
package cn.itcast.spark.day5
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{HashPartitioner, SparkConf}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* Created by root on 2016/5/21.
*/
//(多個zookeeper用,隔開)
//zookeper groupid topics numThreads
//入參:192.168.1.103:2181 g1 test 2
object KafkaWordCount {
val updateFunc = (iter: Iterator[(String, Seq[Int], Option[Int])]) => {
//iter.flatMap(it=>Some(it._2.sum + it._3.getOrElse(0)).map(x=>(it._1,x)))
iter.flatMap { case (x, y, z) => Some(y.sum + z.getOrElse(0)).map(i => (x, i)) }
}
def main(args: Array[String]) {
LoggerLevels.setStreamingLogLevels()
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")
val ssc = new StreamingContext(sparkConf, Seconds(5))
ssc.checkpoint("c://ck2")
//"alog-2016-04-16,alog-2016-04-17,alog-2016-04-18"
//"Array((alog-2016-04-16, 2), (alog-2016-04-17, 2), (alog-2016-04-18, 2))"
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
val data = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap, StorageLevel.MEMORY_AND_DISK_SER)
val words = data.map(_._2).flatMap(_.split(" "))
val wordCounts = words.map((_, 1)).updateStateByKey(updateFunc, new HashPartitioner(ssc.sparkContext.defaultParallelism), true)
wordCounts.print()
ssc.start()
ssc.awaitTermination()
}
}
pom.xml如下
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.itcast.spark</groupId>
<artifactId>hello-spark</artifactId>
<version>1.0</version>
<properties>
<maven.compiler.source>1.7</maven.compiler.source>
<maven.compiler.target>1.7</maven.compiler.target>
<encoding>UTF-8</encoding>
<scala.version>2.10.6</scala.version>
<spark.version>1.6.1</spark.version>
<hadoop.version>2.6.4</hadoop.version>
</properties>
<dependencies>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>${scala.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>${hadoop.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.10</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka_2.10</artifactId>
<version>1.6.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-flume_2.10</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.38</version>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.8.1</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<testSourceDirectory>src/test/scala</testSourceDirectory>
<plugins>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.2.2</version>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
<configuration>
<args>
<arg>-make:transitive</arg>
<arg>-dependencyfile</arg>
<arg>${project.build.directory}/.scala_dependencies</arg>
</args>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.4.3</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>