Kafka整合SparkStream两种方式

迷南。 2023-06-16 13:54 163阅读 0赞

## `Kafka`整合`SparkStream`两种方式 ##

## 官网[http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html][http_spark.apache.org_docs_latest_streaming-kafka-0-8-integration.html] ##

### 方式1：基于`receiver based`的实现 ###

#### 1.1 `kafka zookeeper`环境测试 ####

##### 1.1.1.先启动`kafka`，需要先启动`zk` #####

cd $ZK_HOME/bin
    ./zkServer.sh start
    
    [hadoop@hadoop000 bin]$ jps -m
    6368 Jps -m
    6346 QuorumPeerMain /home/hadoop/app/zookeeper-3.4.5-cdh5.7.0/bin/../conf/zoo.cfg

##### 1.1.2.启动`kafka` #####

cd $KAFKA_HOME/bin
    [hadoop@hadoop000 bin]$ ./kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
    [hadoop@hadoop000 bin]$ jps -m
    6452 Jps -m
    6391 Kafka /home/hadoop/app/kafka_2.11-0.9.0.0/config/server.properties
    6346 QuorumPeerMain /home/hadoop/app/zookeeper-3.4.5-cdh5.7.0/bin/../conf/zoo.cfg

##### 1.1.3.创建`topic` #####

kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 1 --partitions 1 --topic kafka_streaming

##### 1.1.4.查看`topic` #####

kafka-topics.sh --list --zookeeper hadoop000:2181

[hadoop@hadoop000 bin]$ kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 1 --partitions 1 --topic kafka_streaming
    WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
    Created topic "kafka_streaming".
    [hadoop@hadoop000 bin]$ kafka-topics.sh --list --zookeeper hadoop000:2181
    hello_topic
    kafka_streaming
    kafka_streaming_topic
    my-replicated-topic
    my-replicated-topic1
    streamingtopic
    topic_hello

##### 1.1.5 测试 #####

###### 1.1.5.1 通过控制台测试`topic`能够正常生产和消费信息 ######

kafka-console-producer.sh --broker-list hadoop000:9092 --topic kafka_streaming
    
    [hadoop@hadoop000 bin]$ kafka-console-producer.sh --broker-list hadoop000:9092 --topic kafka_streaming
    hellohadoop spark
    zookeeper

###### 1.1.5.2 消费信息 ######

kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic kafka_streaming  --from-beginning
    
    [hadoop@hadoop000 ~]$ kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic kafka_streaming  --from-beginning
    hellohadoop spark
    zookeeper

**至此测试完毕：`kafka和zookeeper`环境ok**

’

#### 1.2 代码开发(本地调试) ####

##### 1.2.1新增`kafka pom`依赖 #####

<dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
          <version>2.2.0</version>
        </dependency>

##### 1.2.2 代码开发 #####

package com.zgw.spark
    
    import org.apache.log4j.{ Level, Logger}
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    /** * Created by Zhaogw&Lss on 2019/11/26. * sparkstreaming对接kafka方式一 */
    object KafkaReceiverWordCount { 
    
      def main(args: Array[String]): Unit = { 
        if (args.length!=4){ 
          System.err.print("Usage:KafkaReceiverWordCount <zkQuorum> <group> <topic> <numThread>")
          System.exit(1)
        }
        val Array(zkQuorum,group,topics,numThread) = args
    
        val sc: SparkConf = new SparkConf().setMaster("local[3]").setAppName("KafkaReceiverWordCount").set("spark.testing.memory", "2147480000")
    
        Logger.getLogger("org").setLevel(Level.ERROR)
        //创建StreamingContext两个参数 SparkConf和batch interval
        val ssc = new StreamingContext(sc, Seconds(5))
    
        val topicMap = topics.split(",").map((_,numThread.toInt)).toMap
    
        val messages = KafkaUtils.createStream(ssc,zkQuorum,group,topicMap)
    
    
        messages.map(_._2).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
    
        ssc.start()
    
        ssc.awaitTermination()
      }
    }

##### 1.2.3 启动参数传递(`IDEA`) #####

#zk地址 消费者组 消费者 线程数
    hadoop000:2181  test  kafka_streaming 1

##### 1.2.4 测试程序 #####

###### 1.2.4.1 启动项目：在控制台的生产者传递消息 ######

[hadoop@hadoop000 bin]$ kafka-console-producer.sh --broker-list hadoop000:9092 --topic 
    A A A B B B C C

###### 1.2.4.2 在控制台的消费者消费消息 ######

[hadoop@hadoop000 ~]$ kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic 
    A A A B B B C C

###### 1.2.4.3 IDEA中消息同时也在消费： ######

-------------------------------------------
    Time: 1574752445000 ms
    -------------------------------------------
    (B,3)
    (C,2)
    (A,3)

#### 1.3 代码修改(生产调试) ####

##### 1.3.1 修改代码(注释如下内容) #####

val sc: SparkConf = new SparkConf()
        /*.setMaster("local[3]").setAppName("KafkaReceiverWordCount").set("spark.testing.memory", "2147480000")*/

##### 1.3.2 打包并发送到`linux`主机上 #####

##### 1.3.3 启动spark程序 #####

spark-submit \
    --class com.zgw.spark.KafkaReceiverWordCount \
    --master local[2] \
    --name KafkaReceiverWordCount \
    --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
    /home/hadoop/lib/SparkTrain-1.0.jar hadoop000:2181  test  kafka_streaming 1

##### 1.3.4 测试 #####

###### 1.3.4.1 `kafka`生产者提交数据 ######

A A A B B B C C

###### 1.3.4.2 程序可以正常消费 ######

-------------------------------------------
    Time: 1574753280000 ms
    -------------------------------------------
    (B,3)
    (A,3)
    (C,2)

### 方式2：基于`Direct Approach`的实现 ###

#### 2.1 代码开发(本地调试) ####

##### 2.1.1 代码开发 #####

package com.zgw.spark.streaming
    
    import kafka.serializer.StringDecoder
    import org.apache.log4j.{ Level, Logger}
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    /** * Created by Zhaogw&Lss on 2019/11/26. * sparkstreaming对接kafka方式二 */
    object KafkaDirectWordCount { 
    
      def main(args: Array[String]): Unit = { 
        if (args.length!=2){ 
          System.err.print("Usage:KafkaReceiverWordCount <brokers> <topic> ")
          System.exit(1)
        }
        /*val Array(hostname,port) = args*/
    
        val Array(brokers,topics) = args
    
        val sc: SparkConf = new SparkConf().setMaster("local[3]").setAppName("KafkaReceiverWordCount").set("spark.testing.memory", "2147480000")
    
        Logger.getLogger("org").setLevel(Level.ERROR)
        //创建StreamingContext两个参数 SparkConf和batch interval
        val ssc = new StreamingContext(sc, Seconds(5))
    
        val topicSet = topics.split(",").toSet
    
        val kafkaParams = Map[String,String]("metadata.broker.list" -> brokers)
    //核心代码：创建createDirectStream并传入泛型[String,String,StringDecoder,StringDecoder]
        val messages = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topicSet)
    
    
        messages.map(_._2).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
    
        ssc.start()
    
        ssc.awaitTermination()
      }
    }

##### 2.1.2 `IDEA`配置启动参数：`hadoop000:9092 kafka_streaming` #####

##### 2.1.3 测试程序 #####

###### 2.1.3.1在生产者中传入： ######

A A A B B B C C

###### 2.1.3.2 `spark_streaming`消费 ######

-------------------------------------------
    Time: 1574758320000 ms
    -------------------------------------------
    (B,3)
    (C,2)
    (A,3)

#### 2.2 代码修改(生产环境) ####

##### 2.2.1 修改代码： #####

val sc: SparkConf = new SparkConf()
     /*.setMaster("local[3]").setAppName("KafkaReceiverWordCount").set("spark.testing.memory", "2147480000")*/

##### 2.2.2 打包并传到`linux`生产 #####

##### 2.2.3 提交启动 #####

spark-submit \
    --class com.zgw.spark.KafkaDirectWordCount \
    --master local[2] \
    --name KafkaDirectWordCount \
    --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
    /home/hadoop/lib/SparkTrain-1.0.jar hadoop000:9092 kafka_streaming

##### 2.2.4 测试 #####

###### 2.2.4.1 发送数据 ######

\[hadoop@hadoop000 bin\]$ kafka-console-producer.sh --broker-list hadoop000:9092 --topic kafka\_streaming

`hadoop hadoop spark`

###### 2.2.4.2 消费数据 ######

-------------------------------------------
    Time: 1574759390000 ms
    -------------------------------------------
    
    -------------------------------------------
    Time: 1574759395000 ms
    -------------------------------------------
    (spark,1)
    (hadoop,2)

[http_spark.apache.org_docs_latest_streaming-kafka-0-8-integration.html]: http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html