Spark 系列（十六）—— Spark Streaming 整合 Kafka

一、版本说明

Spark 针对 Kafka 的不同版本，提供了两套整合方案：spark-streaming-kafka-0-8 和 spark-streaming-kafka-0-10，其主要区别如下：

spark-streaming-kafka-0-8 spark-streaming-kafka-0-10
Kafka 版本 0.8.2.1 or higher 0.10.0 or higher
AP 状态 Deprecated
从 Spark 2.3.0 版本开始，Kafka 0.8 支持已被弃用		 Stable(稳定版)
语言支持 Scala, Java, Python Scala, Java
Receiver DStream Yes No
Direct DStream Yes Yes
SSL / TLS Support No Yes
Offset Commit API(偏移量提交) No Yes
Dynamic Topic Subscription
(动态主题订阅)		 No Yes

本文使用的 Kafka 版本为 kafka_2.12-2.2.0，故采用第二种方式进行整合。

二、项目依赖

项目采用 Maven 进行构建，主要依赖如下：

<properties>
    <scala.version>2.12</scala.version>
</properties>

<dependencies>
    <!-- Spark Streaming-->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_${scala.version}</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <!-- Spark Streaming 整合 Kafka 依赖-->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-10_${scala.version}</artifactId>
        <version>2.4.3</version>
    </dependency>
</dependencies>

完整源码见本仓库：spark-streaming-kafka

三、整合Kafka

通过调用 KafkaUtils 对象的 createDirectStream 方法来创建输入流，完整代码如下：

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * spark streaming 整合 kafka
  */
object KafkaDirectStream {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setAppName("KafkaDirectStream").setMaster("local[2]")
    val streamingContext = new StreamingContext(sparkConf, Seconds(5))

    val kafkaParams = Map[String, Object](
      /*
       * 指定 broker 的地址清单，清单里不需要包含所有的 broker 地址，生产者会从给定的 broker 里查找其他 broker 的信息。
       * 不过建议至少提供两个 broker 的信息作为容错。
       */
      "bootstrap.servers" -> "hadoop001:9092",
      /*键的序列化器*/
      "key.deserializer" -> classOf[StringDeserializer],
      /*值的序列化器*/
      "value.deserializer" -> classOf[StringDeserializer],
      /*消费者所在分组的 ID*/
      "group.id" -> "spark-streaming-group",
      /*
       * 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理:
       * latest: 在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
       * earliest: 在偏移量无效的情况下，消费者将从起始位置读取分区的记录
       */
      "auto.offset.reset" -> "latest",
      /*是否自动提交*/
      "enable.auto.commit" -> (true: java.lang.Boolean)
    )

    /*可以同时订阅多个主题*/
    val topics = Array("spark-streaming-topic")
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      /*位置策略*/
      PreferConsistent,
      /*订阅主题*/
      Subscribe[String, String](topics, kafkaParams)
    )

    /*打印输入流*/
    stream.map(record => (record.key, record.value)).print()

    streamingContext.start()
    streamingContext.awaitTermination()
  }
}

3.1 ConsumerRecord

这里获得的输入流中每一个 Record 实际上是 ConsumerRecord<K, V> 的实例，其包含了 Record 的所有可用信息，源码如下：

public class ConsumerRecord<K, V> {

    public static final long NO_TIMESTAMP = RecordBatch.NO_TIMESTAMP;
    public static final int NULL_SIZE = -1;
    public static final int NULL_CHECKSUM = -1;

    /*主题名称*/
    private final String topic;
    /*分区编号*/
    private final int partition;
    /*偏移量*/
    private final long offset;
    /*时间戳*/
    private final long timestamp;
    /*时间戳代表的含义*/
    private final TimestampType timestampType;
    /*键序列化器*/
    private final int serializedKeySize;
    /*值序列化器*/
    private final int serializedValueSize;
    /*值序列化器*/
    private final Headers headers;
    /*键*/
    private final K key;
    /*值*/
    private final V value;
    .....
}

3.2 生产者属性

在示例代码中 kafkaParams 封装了 Kafka 消费者的属性，这些属性和 Spark Streaming 无关，是 Kafka 原生 API 中就有定义的。其中服务器地址、键序列化器和值序列化器是必选的，其他配置是可选的。其余可选的配置项如下：

1. fetch.min.byte

消费者从服务器获取记录的最小字节数。如果可用的数据量小于设置值，broker 会等待有足够的可用数据时才会把它返回给消费者。

2. fetch.max.wait.ms

broker 返回给消费者数据的等待时间。

3. max.partition.fetch.bytes

分区返回给消费者的最大字节数。

4. session.timeout.ms

消费者在被认为死亡之前可以与服务器断开连接的时间。

5. auto.offset.reset

该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：

  • latest(默认值) ：在偏移量无效的情况下，消费者将从其启动之后生成的最新的记录开始读取数据；
  • earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录。

6. enable.auto.commit

是否自动提交偏移量，默认值是 true,为了避免出现重复数据和数据丢失，可以把它设置为 false。

7. client.id

客户端 id，服务器用来识别消息的来源。

8. max.poll.records

单次调用 poll() 方法能够返回的记录数量。

9. receive.buffer.bytes 和 send.buffer.byte

这两个参数分别指定 TCP socket 接收和发送数据包缓冲区的大小，-1 代表使用操作系统的默认值。

3.3 位置策略

Spark Streaming 中提供了如下三种位置策略，用于指定 Kafka 主题分区与 Spark 执行程序 Executors 之间的分配关系：

  • PreferConsistent : 它将在所有的 Executors 上均匀分配分区；
  • PreferBrokers : 当 Spark 的 Executor 与 Kafka Broker 在同一机器上时可以选择该选项，它优先将该 Broker 上的首领分区分配给该机器上的 Executor；
  • PreferFixed : 可以指定主题分区与特定主机的映射关系，显示地将分区分配到特定的主机，其构造器如下：
@Experimental
def PreferFixed(hostMap: collection.Map[TopicPartition, String]): LocationStrategy =
  new PreferFixed(new ju.HashMap[TopicPartition, String](hostMap.asJava))

@Experimental
def PreferFixed(hostMap: ju.Map[TopicPartition, String]): LocationStrategy =
  new PreferFixed(hostMap)

3.4 订阅方式

Spark Streaming 提供了两种主题订阅方式，分别为 Subscribe 和 SubscribePattern。后者可以使用正则匹配订阅主题的名称。其构造器分别如下：

/**
  * @param 需要订阅的主题的集合
  * @param Kafka 消费者参数
  * @param offsets(可选): 在初始启动时开始的偏移量。如果没有，则将使用保存的偏移量或 auto.offset.reset 属性的值
  */
def Subscribe[K, V](
    topics: ju.Collection[jl.String],
    kafkaParams: ju.Map[String, Object],
    offsets: ju.Map[TopicPartition, jl.Long]): ConsumerStrategy[K, V] = { ... }

/**
  * @param 需要订阅的正则
  * @param Kafka 消费者参数
  * @param offsets(可选): 在初始启动时开始的偏移量。如果没有，则将使用保存的偏移量或 auto.offset.reset 属性的值
  */
def SubscribePattern[K, V](
    pattern: ju.regex.Pattern,
    kafkaParams: collection.Map[String, Object],
    offsets: collection.Map[TopicPartition, Long]): ConsumerStrategy[K, V] = { ... }

在示例代码中，我们实际上并没有指定第三个参数 offsets，所以程序默认采用的是配置的 auto.offset.reset 属性的值 latest，即在偏移量无效的情况下，消费者将从其启动之后生成的最新的记录开始读取数据。

3.5 提交偏移量

在示例代码中，我们将 enable.auto.commit 设置为 true，代表自动提交。在某些情况下，你可能需要更高的可靠性，如在业务完全处理完成后再提交偏移量，这时候可以使用手动提交。想要进行手动提交，需要调用 Kafka 原生的 API :

  • commitSync: 用于异步提交；
  • commitAsync：用于同步提交。

具体提交方式可以参见：Kafka 消费者详解

四、启动测试

4.1 创建主题

1. 启动Kakfa

Kafka 的运行依赖于 zookeeper，需要预先启动，可以启动 Kafka 内置的 zookeeper，也可以启动自己安装的：

# zookeeper启动命令
bin/zkServer.sh start

# 内置zookeeper启动命令
bin/zookeeper-server-start.sh config/zookeeper.properties

启动单节点 kafka 用于测试：

# bin/kafka-server-start.sh config/server.properties

2. 创建topic

# 创建用于测试主题
bin/kafka-topics.sh --create                     --bootstrap-server hadoop001:9092                     --replication-factor 1                     --partitions 1                      --topic spark-streaming-topic

# 查看所有主题
 bin/kafka-topics.sh --list --bootstrap-server hadoop001:9092

3. 创建生产者

这里创建一个 Kafka 生产者，用于发送测试数据：

bin/kafka-console-producer.sh --broker-list hadoop001:9092 --topic spark-streaming-topic

4.2 本地模式测试

这里我直接使用本地模式启动 Spark Streaming 程序。启动后使用生产者发送数据，从控制台查看结果。

从控制台输出中可以看到数据流已经被成功接收，由于采用 kafka-console-producer.sh 发送的数据默认是没有 key 的，所以 key 值为 null。同时从输出中也可以看到在程序中指定的 groupId 和程序自动分配的 clientId。

参考资料

  1. https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

更多大数据系列文章可以参见 GitHub 开源项目： 大数据入门指南

原文地址：https://www.cnblogs.com/heibaiying/p/11359594.html

时间： 2024-11-09 01:59:49

Spark 系列（十六）—— Spark Streaming 整合 Kafka的相关文章

Spark Streaming整合Kafka

0)摘要 主要介绍了Spark Streaming整合Kafka,两种整合方式:Receiver-based和Direct方式.这里使用的是Kafka broker version 0.8.2.1,官方文档地址:(http://spark.apache.org/docs/2.2.0/streaming-kafka-0-8-integration.html). 1)Kafka准备 启动zookeeper ./zkServer.sh start 启动kafka ./kafka-server-star

【spark系列3】spark开发简单指南

分布式数据集创建之textFile 文本文件的RDDs能够通过SparkContext的textFile方法创建,该方法接受文件的URI地址(或者机器上的文件本地路径,或者一个hdfs://, sdn://,kfs://,其他URI).这里是一个调用样例:scala> val distFile = sc.textFile("data.txt")distFile: spark.RDD[String] = [email protected] 分布式数据集操作之转换和动作 分布式数据集

Spark（十） -- Spark Streaming API编程

本文测试的Spark版本是1.3.1 Spark Streaming编程模型: 第一步: 需要一个StreamingContext对象,该对象是Spark Streaming操作的入口 ,而构建一个StreamingContext对象需要两个参数: 1.SparkConf对象:该对象是配置Spark 程序设置的,例如集群的Master节点,程序名等信息 2.Seconds对象:该对象设置了StreamingContext多久读取一次数据流 第二步: 构建好入口对象之后,直接调用该入口的方法读取各

Spark 系列（六）—— 累加器与广播变量

一.简介 在 Spark 中,提供了两种类型的共享变量:累加器 (accumulator) 与广播变量 (broadcast variable): 累加器:用来对信息进行聚合,主要用于累计计数等场景: 广播变量:主要用于在节点间高效分发大对象. 二.累加器 这里先看一个具体的场景,对于正常的累计求和,如果在集群模式中使用下面的代码进行计算,会发现执行结果并非预期: var counter = 0 val data = Array(1, 2, 3, 4, 5) sc.parallelize(dat

Spark系列(十)TaskSchedule工作原理

工作原理图 源码分析: 1.submitTasks 在submitTasks方法中最后调用backend.reviveOffers()进行下一步的task调度分配 1  override def submitTasks(taskSet: TaskSet) { 2      val tasks = taskSet.tasks 3      logInfo("Adding task set " + taskSet.id + " with " + tasks.length

SpringBoot系列十二：SpringBoot整合 Shiro

1.概念:SpringBoot 整合 Shiro 2.具体内容 Shiro 是现在最为流行的权限认证开发框架,与它起名的只有最初的 SpringSecurity(这个开发框架非常不好用,但是千万不要 以为 SpringSecurity 没有用处,它在 SpringCloud 阶段将发挥重大的作用).但是现在如果要想整合 Shiro 开发框架有一点很遗憾, SpringBoot 没有直接的配置支持,它不像整合所谓的 Kafka.Redis.DataSource,也就是说如果要想整合 Shiro 开

Cocos2d-x 3.x 图形学渲染系列十六

笔者介绍:姜雪伟,IT公司技术合伙人,IT高级讲师,CSDN社区专家,特邀编辑,畅销书作者,国家专利发明人;已出版书籍:<手把手教你架构3D游戏引擎>电子工业出版社和<Unity3D实战核心技术详解>电子工业出版社等. 每个引擎都有自己的处理Shader类,Cocos使用的是GLProgram类,之所以定义GLProgram类,是因为在引擎中需要有一个类管理模型的信息和矩阵信息声明.在GLProgram类中定义了模型顶点的属性,这些属性在加载模型时,用于解释模型文件内容时用于做属性

Hadoop运维记录系列(十六)

应了一个国内某电信运营商集群恢复的事,集群故障很严重,做了HA的集群Namenode挂掉了.具体过程不详,但是从受害者的只言片语中大概回顾一下历史的片段. Active的namenode元数据硬盘满了,满了,满了...上来第一句话就如雷贯耳. 运维人员发现硬盘满了以后执行了对active namenode的元数据日志执行了 echo "" > edit_xxxx-xxxx...第二句话如五雷轰顶. 然后发现standby没法切换,切换也没用,因为standby的元数据和日志是5月

Java设计模式菜鸟系列(十六)原型模式建模与实现

转载请注明出处:http://blog.csdn.net/lhy_ycu/article/details/39997337 原型模式(Prototype):该模式的思想就是将一个对象作为原型,对其进行复制.克隆,产生一个和原对象类似的新对象.而这里的复制有两种:浅复制.深复制. 浅复制:将一个对象复制后,基本数据类型的变量都会重新创建,而引用类型,指向的还是原对象所指向的. 深复制:将一个对象复制后,不论是基本数据类型还有引用类型,都是重新创建的.简单来说,就是深复制进行了完全彻底的复制,而浅复