SparkStreaming 整合kafka实例

悠悠 2023-05-31 06:37 71阅读 0赞

## kafka基础 ##

**核心概念**  
下面介绍Kafka相关概念,以便运行下面实例的同时，更好地理解Kafka.

*  Broker  
    Kafka集群包含一个或多个服务器，这种服务器被称为broker
 *  Topic  
    每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
 *  Partition  
    Partition是物理上的概念，每个Topic包含一个或多个Partition.
 *  Producer  
    负责发布消息到Kafka broker
 *  Consumer  
    消息消费者，向Kafka broker读取消息的客户端。
 *  Consumer Group  
    每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）

### 测试简单实例 ###

接下来在Ubuntu系统环境下测试简单的实例。按顺序执行如下命令：

进入kafka所在的目录

cd /usr/local/kafka
    bin/zookeeper-server-start.sh config/zookeeper.properties

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]

命令执行后不会返回Shell命令输入状态,zookeeper就会按照默认的配置文件启动服务,请千万不要关闭当前终端.启动新的终端，输入如下命令：

cd /usr/local/kafka
    bin/kafka-server-start.sh config/server.properties

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 1]

kafka服务端就启动了,请千万不要关闭当前终端。启动另外一个终端,输入如下命令:

cd /usr/local/kafka
    bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic zhaogw

topic是发布消息发布的category,以单节点的配置创建了一个叫zhaogw的topic.可以用list列出所有创建的topics,来查看刚才创建的主题是否存在。

bin/kafka-topics.sh --list --zookeeper localhost:2181

![在这里插入图片描述][20191023094655471.png]  
可以在结果中查看到zhaogw这个topic存在。接下来用producer生产点数据：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic zhaogw

![在这里插入图片描述][20191023094807660.png]  
并尝试输入如下信息：

hello world
    hello hadoop

然后再次开启新的终端或者直接按CTRL+C退出。然后使用consumer来接收数据,输入如下命令：

cd /usr/local/kafka
    bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zhaogw   --from-beginning

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 2]

便可以看到刚才产生的三条信息。说明kafka安装成功。

### 接下来用SparkStreaming消费kafka的数据 ###

package com.zgw.spark.streaming
    
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.dstream.{ DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    
    /** * Created by Zhaogw&Lss on 2019/10/22. */
    object Sparkstreaming_Kafka { 
      def main(args: Array[String]): Unit = { 
    
            var sparkConf =new SparkConf().setMaster("local[*]").setAppName("SparkStream").set("spark.testing.memory", "2147480000")
    
            //分析环境对象以及采集周期
            val streamContext = new StreamingContext(sparkConf,Seconds(25))
    
          /* val inputFile = "hdfs://192.168.181.128:8020/spark/"*/
    
           //kafka Stream
        val kafkaDsream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(
          streamContext,
          "dblab-VirtualBox:2181",
          "zhaogw",
          Map("zhaogw" -> 3)
        )
    
            //将采集数据进行分解
            val dStream: DStream[String] = kafkaDsream.flatMap(t=> t._2.split(" "))
    
            //将数据进行结构转变
            val map: DStream[(String, Int)] = dStream.map((_,1))
            //聚合处理
            val key: DStream[(String, Int)] = map.reduceByKey(_+_)
            //结果打印
            key.print()
            //启动采集器
            streamContext.start()
            //等待采集器执行
            streamContext.awaitTermination()
    
      }
    
    }

**核心代码**

//kafka Stream
        val kafkaDsream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(
          streamContext,
          "dblab-VirtualBox:2181",
          "zhaogw",
          Map("zhaogw" -> 3)
        )

查看该方法的源码

def createStream(
          ssc: StreamingContext,
          zkQuorum: String,
          groupId: String,
          topics: Map[String, Int],
          storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
        )

这里使用到了`kafka`的工具类 `KafkaUtils`，第一个参数是`StreamingContext`对象，第二个参数是zk所在的主机名(与hosts文件中的配置对应)，第三个参数是groupId，第四个参数是topics。

**启动项目，并在linux主机发送两条消息**

hello zzzz
    hello heihei

![在这里插入图片描述][2019102309585312.png]  
然后再IDEA中就可以看见消费的数据了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]: /images/20230531/d6d0e640275445128a5763752b65f90a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 1]: /images/20230531/05de8f5070e54581bc4ad9823cd0238b.png
[20191023094655471.png]: /images/20230531/b4abb1c3235343f6abebf7c17bb137b9.png
[20191023094807660.png]: /images/20230531/ee7ef0c9a6d34d71b8a961ed6f57af3f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 2]: /images/20230531/61a32c6b294c42af9dda82c90b5b96bd.png
[2019102309585312.png]: /images/20230531/b0ab838bf6a1466aaa660744d12bc5e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 3]: /images/20230531/fe7f985b144549548767e02705ea38fa.png