Spark Streaming中读取数据的两种方式：基于Receiver的方法，基于Direct方法

蔚落 2021-11-01 12:06 251阅读 0赞

# 基于Receiver的方法——基于Receiver的方法 #

此方法使用Receiver接收数据。 Receiver是使用Kafka高级消费者API实现的。 与所有接收器一样，从Kafka通过Receiver接收的数据存储在Spark执行器中，然后由Spark Streaming启动的作业处理数据。

但是，在默认配置下，此方法可能会在失败时丢失数据（请参阅接收器可靠性。为确保零数据丢失，您必须在Spark Streaming中另外启用Write Ahead Logs（在Spark 1.2中引入）。这将同步保存所有收到的Kafka 将数据写入分布式文件系统（例如HDFS）上的预写日志，以便在发生故障时可以恢复所有数据。有关“预读日志”的更多详细信息，请参阅流式编程指南中的“部署”部分。

## 使用语法： ##

import org.apache.spark.streaming.kafka._
    
     val kafkaStream = KafkaUtils.createStream(streamingContext,
         [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

## 关键点 ##

1.  **Kafka中的主题分区与Spark Streaming中生成的RDD分区无关。** 因此，增加KafkaUtils.createStream（）中特定于主题的分区的数量只会增加使用单个接收器中使用的主题的线程数。 它不会增加Spark在处理数据时的并行性。
2.  可以使用不同的组和主题创建多个Kafka输入DStream，以使用多个接收器并行接收数据。
3.  如果已使用复制文件系统（如HDFS）启用了“预读日志”，则已在日志中复制接收的数据。

# 基于Direct方法——Direct Approach (No Receivers) #

Spark 1.3中引入了这种新的无接收器“直接”方法，以确保更强大的端到端保证。 这种方法不是使用接收器来接收数据，而是定期向Kafka查询每个主题+分区中的最新偏移量，并相应地定义要在每个批次中处理的偏移量范围。 当启动处理数据的作业时，Kafka的简单消费者API用于从Kafka读取定义的偏移范围（类似于从文件系统读取的文件）。 请注意，此功能是在Spark 1.3中为Scala和Java API引入的，在Python 1.4中为Python API引入。

## 优点 ##

1.  简化的并行性：无需创建多个输入Kafka流并将它们联合起来。 使用directStream，**Spark Streaming将创建与要使用的Kafka分区一样多的RDD分区，这些分区将并行地从Kafka读取数据**。 因此，Kafka和RDD分区之间存在一对一的映射，这更容易理解和调整。
2.  效率：在第一种方法中实现零数据丢失需要将数据存储在Write Ahead Log中，这进一步复制了数据。 这实际上是低效的，因为数据有效地被复制两次 - 一次由Kafka复制，第二次由Write Ahead Log复制。 **第二种方法消除了问题，因为没有接收器，因此不需要Write Ahead Logs。** 只要您有足够的Kafka保留，就可以从Kafka恢复消息。
3.  完全一次的语义：第一种方法使用Kafka的高级API在Zookeeper中存储消耗的偏移量。传统上，这是从Kafka使用数据的方式。虽然这种方法（与预写日志结合使用）可以确保零数据丢失（即至少一次语义），但某些记录在某些故障下可能会被消耗两次的可能性很小。这是因为Spark Streaming可靠接收的数据与Zookeeper跟踪的偏移之间存在不一致。\*\*因此，在第二种方法中，我们使用Zookeeper的简单Kafka API。 Spark Streaming在其检查点内跟踪偏移量。这消除了Spark Streaming和Zookeeper / Kafka之间的不一致，\*\*因此尽管出现故障，Spark Streaming也会有效地接收每条记录一次。为了实现输出结果的一次性语义，将数据保存到外部数据存储的输出操作必须是幂等的，或者是保存结果和偏移的原子事务

## 使用语法 ##

import org.apache.spark.streaming.kafka._
    
     val directKafkaStream = KafkaUtils.createDirectStream[
         [key class], [value class], [key decoder class], [value decoder class] ](
         streamingContext, [map of Kafka parameters], [set of topics to consume])

## 如果希望基于Zookeeper的Kafka监视工具显示流应用程序的进度，可以使用此命令自行更新Zookeeper。 ##

// Hold a reference to the current offset ranges, so it can be used downstream
     var offsetRanges = Array.empty[OffsetRange]
    
     directKafkaStream.transform { rdd =>
       offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
       rdd
     }.map {
               ...
     }.foreachRDD { rdd =>
       for (o <- offsetRanges) {
         println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
       }
       ...
     }

原文：[http://spark.apache.org/docs/2.2.0/streaming-kafka-0-8-integration.html][http_spark.apache.org_docs_2.2.0_streaming-kafka-0-8-integration.html]

[http_spark.apache.org_docs_2.2.0_streaming-kafka-0-8-integration.html]: http://spark.apache.org/docs/2.2.0/streaming-kafka-0-8-integration.html