Spark Streaming + Kafka 构造指南（Kafka broker version 0.10.0 or higher）

向右看齐 2023-07-11 11:26 8阅读 0赞

**Table of Contents**

依赖

创建一个直接流

偏移量策略

消费者的策略

Creating an RDD

获得offsets

存储offsets

Checkpoints

Kafka itself

Your own data store

SSL / TLS

部署

--------------------

Kafka 0.10 的 Spark 流集成在设计上类似于0.8直接流方法。它提供了简单的并行性，Kafka 分区和 Spark 分区之间的1:1对应，以及对偏移量和元数据的访问。但是，因为新的集成使用新的 Kafka 消费者 API 而不是简单的 API，所以在使用上有显著的不同。这个版本的集成被标记为实验性的，所以API可能会发生变化。

# 依赖 #

对于使用 SBT/Maven 项目定义的 Scala/Java 应用程序，将您的流应用程序链接到以下工件

groupId = org.apache.spark
    artifactId = spark-streaming-kafka-0-10_2.11
    version = 2.1.1

# 创建一个直接流 #

注意，导入的名称空间包括版本 org.apache.spark.stream .kafka010

import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.kafka.common.serialization.StringDeserializer
    import org.apache.spark.streaming.kafka010._
    import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
    import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
    
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "use_a_separate_group_id_for_each_stream",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    
    val topics = Array("topicA", "topicB")
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    
    stream.map(record => (record.key, record.value))

流中的每一条都是一个 [ConsumerRecord][]

有关 kafkaParams, see [Kafka consumer config docs][]。如果您的 Spark 批处理持续时间大于默认的 Kafka 心跳会话超时(30秒)，请增加 heartbeat.interval.ms and session.timeout.ms。对于大于5分钟的批，这将需要更改 kafka broker 的 group.max.session.timeout.ms。注意，该示例将 enable.auto.commit 设置为false，有关讨论请参阅下面的存储偏移量。

# 偏移量策略 #

新的 Kafka 消费者 API 将预取消息到缓冲区。因此，出于性能方面的考虑，Spark 集成将缓存的使用者保留在执行器中(而不是为每个批处理重新创建它们)，并且更喜欢将分区安排在拥有适当使用者的主机位置上，这一点非常重要。

在大多数情况下，你应该使用定位策略。如上面所示，首选一致性。这将在可用的执行者之间均匀地分配分区。如果您的执行者与 Kafka 代理位于相同的主机上，那么使用 PreferBrokers，它更愿意为该分区在 Kafka leader 上调度分区。最后，如果分区之间的负载有明显的倾斜，那么使用 PreferFixed。这允许您指定分区到主机的显式映射(任何未指定的分区将使用一致的位置)。

使用者缓存的默认最大大小为64。如果您希望处理多于(64 \*执行器数量)Kafka分区，您可以通过 `spark.streaming.kafka.consumer.cache.maxCapacity `来更改该设置。

缓存由 topic partition和 group.id 键控。所以每个 createDirectStream 的调用使用一个单独的 group.id。

# 消费者的策略 #

新的 Kafka 消费者 API 有许多不同的方式来指定主题，其中一些需要大量的对象实例化后的设置。客户策略提供了一个抽象，允许 Spark 即使在从检查点重新启动之后也可以获得正确配置的客户。

如上所示，ConsumerStrategies.Subscribe 允许您订阅固定的主题集合。SubscribePattern 允许您使用正则表达式来指定主题。注意，与0.8集成不同，使用 Subscribe 或 SubscribePattern 应该在运行流期间响应添加分区。最后，Assign 允许您指定一个固定的分区集合。这三种策略都有重载的构造函数，允许您指定特定分区的起始偏移量。

如果您有特定的消费者设置需求，而上面的选项不能满足这些需求，那么 ConsumerStrategy 是一个可以扩展的公共类。

# Creating an RDD #

如果您有一个更适合批处理的用例，那么您可以为定义的偏移范围创建一个RDD。

// Import dependencies and create kafka params as in Create Direct Stream above
    
    val offsetRanges = Array(
      // topic, partition, inclusive starting offset, exclusive ending offset
      OffsetRange("test", 0, 0, 100),
      OffsetRange("test", 1, 0, 100)
    )
    
    val rdd = KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent)

请注意，您不能使用 preferbroker，因为没有流，就没有一个驱动端使用者为您自动查找代理元数据。如果需要，可以在自己的元数据查找中使用 PreferFixed。

# 获得offsets #

stream.foreachRDD { rdd =>
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd.foreachPartition { iter =>
        val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
        println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
      }
    }

注意，对 HasOffsetRanges 的类型转换只有在对 createDirectStream 的结果调用的第一个方法中完成时才会成功，而不是在随后的方法链中.

请注意，RDD 分区和 Kafka 分区之间的一对一映射不会在任何洗牌或重分区的方法(例如reduceByKey()或window())之后仍然存在。

# 存储offsets #

失败情况下的 Kafka 交付语义取决于如何以及何时存储偏移量。spark 输出操作至少一次。因此，如果您想要完全的一次语义，您必须在幂等的输出之后存储偏移量，或者将偏移量与输出一起存储在原子事务中。有了这个集成，您有3个选项，为了增加可靠性(和代码复杂性)，来存储偏移量。

## Checkpoints ##

如果启用了spark检查点，则偏移量将存储在检查点中。这很容易实现，但也有缺点。你的输出操作必须是幂等的，因为你会得到重复的输出;交易不是一个选项。此外，如果应用程序代码发生了更改，则无法从检查点恢复。对于计划中的升级，您可以通过在运行旧代码的同时运行新代码来缓解这种情况(因为无论如何输出都需要是幂等的，它们不应该冲突)。但是对于需要更改代码的计划外故障，除非您有其他方法来识别已知的良好起始偏移量，否则您将丢失数据。

## Kafka itself ##

Kafka 有一个偏移量提交 API，它将偏移量存储在一个特殊的 Kafka 主题中。默认情况下，新使用者将定期自动提交补偿。这几乎肯定不是您想要的，因为由使用者成功轮询的消息可能尚未导致 Spark 输出操作，从而导致语义未定义。

这就是为什么上面的流示例将“enable.auto.commit”设置为false。但是，您可以使用 commitAsync API 在知道您的输出已被存储后将偏移量提交给 Kafka。与检查点相比，Kafka 的优点是无论应用程序代码如何更改，它都是一个持久的存储。然而，Kafka 不是事务性的，所以你的输出必须是幂等的。

stream.foreachRDD { rdd =>
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    
      // some time later, after outputs have completed
      stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
    }

与 HasOffsetRanges 一样，只有在 createDirectStream 的结果上调用时，转换到 CanCommitOffsets 才会成功，而不是在转换之后。commitAsync  调用是线程安全的，但如果需要有意义的语义，则必须在输出之后执行。

## Your own data store ##

对于支持事务的数据存储，将偏移量保存在与结果相同的事务中可以使两者保持同步，即使在出现故障的情况下也是如此。如果您小心地检测重复或跳过的偏移范围，回滚事务可以防止重复或丢失的消息影响结果。

这就给出了完全一次的语义。甚至对于聚合产生的输出也可以使用这种策略，因为聚合通常很难实现幂等性。

// The details depend on your data store, but the general idea looks like this
    
    // begin from the the offsets committed to the database
    val fromOffsets = selectOffsetsFromYourDatabase.map { resultSet =>
      new TopicPartition(resultSet.string("topic"), resultSet.int("partition")) -> resultSet.long("offset")
    }.toMap
    
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets)
    )
    
    stream.foreachRDD { rdd =>
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    
      val results = yourCalculation(rdd)
    
      // begin your transaction
    
      // update results
      // update offsets where the end of existing offsets matches the beginning of this batch of offsets
      // assert that offsets were updated correctly
    
      // end your transaction
    }

# SSL / TLS #

新的 Kafka 使用者支持 SSL。要启用它，请在传递到 createDirectStream / createRDD 之前适当地设置 kafkaParams。

注意，这只适用于 Spark 和 Kafka 代理之间的通信；您仍然需要单独负责保护 Spark 节点间通信。

val kafkaParams = Map[String, Object](
      // the usual params, make sure to change the port in bootstrap.servers if 9092 is not TLS
      "security.protocol" -> "SSL",
      "ssl.truststore.location" -> "/some-directory/kafka.client.truststore.jks",
      "ssl.truststore.password" -> "test1234",
      "ssl.keystore.location" -> "/some-directory/kafka.client.keystore.jks",
      "ssl.keystore.password" -> "test1234",
      "ssl.key.password" -> "test1234"
    )

# 部署 #

与任何 Spark 应用程序一样，Spark -submit 用于启动应用程序。

对于 Scala 和 Java 应用程序，如果您使用SBT或Maven进行项目管理，那么可以将spark-streaming-kafka-0-10\_2.11及其依赖项打包到应用程序JAR中。确保Spark -core\_2.11和Spark -streaming\_2.11被标记为提供的依赖项，就像它们在Spark安装中已经存在一样。然后使用spark-submit启动您的应用程序

原文地址：[http://spark.apache.org/docs/2.1.1/streaming-kafka-0-10-integration.html][http_spark.apache.org_docs_2.1.1_streaming-kafka-0-10-integration.html]

[ConsumerRecord]: http://kafka.apache.org/0100/javadoc/org/apache/kafka/clients/consumer/ConsumerRecord.html
[Kafka consumer config docs]: http://kafka.apache.org/documentation.html#newconsumerconfigs
[http_spark.apache.org_docs_2.1.1_streaming-kafka-0-10-integration.html]: http://spark.apache.org/docs/2.1.1/streaming-kafka-0-10-integration.html