Structured Streaming + Kafka Integration Guide 结构化流+Kafka集成指南 (Kafka broker version 0.10.0 or higher)

时间:2022-10-12 17:08:04

 用于Kafka 0.10的结构化流集成从Kafka读取数据并将数据写入到Kafka。

1. Linking

对于使用SBT/Maven项目定义的Scala/Java应用程序,用以下工件artifact连接你的应用程序:

Structured Streaming + Kafka Integration Guide 结构化流+Kafka集成指南 (Kafka broker version 0.10.0 or higher)

对于Python应用程序,你需要在部署应用程序时添加上面的库及其依赖关系。查看Deploying子节点。

2. Reading Data from Kafka 从Kafka读取数据

2.1 Creating a Kafka Source for Streaming Queries 为流式查询创建一个Kafka来源

Structured Streaming + Kafka Integration Guide 结构化流+Kafka集成指南 (Kafka broker version 0.10.0 or higher)

2.2 Creating a Kafka Source for Batch Queries 为批处理查询创建一个Kafka来源

如果你有一个更适合用于批处理的用例,你可以为定义的偏移量范围创建一个Dataset/DataFrame。

Structured Streaming + Kafka Integration Guide 结构化流+Kafka集成指南 (Kafka broker version 0.10.0 or higher)

来源的每一行有以下格式:

Column Type
key binary
value binary
topic string
partition int
offset long
timestamp long
timestampType int

对于批处理和流式查询,必须为Kafka来源设置以下选项。

Option value meaning
assign json string {"topicA":[0,1],"topicB":[2,4]} 要使用特定TopicPartition。对于Kafka来源,只能指定“assign”,"subscribe"或者“subscribePattern”选项中的一个。
subscribe A comma-separated list of topics 要订阅的主题列表.对于Kafka来源,只能指定“assign”,"subscribe"或者“subscribePattern”选项中的一个。
subscribePattern Java regex string 用于订阅主题的格式。对于Kafka来源,只能指定“assign”,"subscribe"或者“subscribePattern”选项中的一个。
kafka.bootstrap.servers A comma-separated list of host:port Kafka "bootstrap.servers" 配置。

以下配置是可选的:

Option value default query type meaning
startingOffsets "earliest", "latest" (streaming only), or json string """ {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}} """ "latest"用于流, "earliest" 用于批量

streaming and batch

流和批量

查询开始时的起始点,可以是最早偏移量的“earliest”,也可以是最近偏移量的“latest”,也可以是指定每个TopicPartition起始偏移量的json字符串。在Json中,-2作为偏移量可以用来指最早的,-1指最新的。注意:对于批量查询,不允许使用最新(隐式或在json中使用-1)。对于流式查询,这仅适用于新查询开始时的情况,并且恢复将始终从查询停止的地方开始。在查询期间新发现的分区将从earliest开始。
endingOffsets latest or json string {"topicA":{"0":23,"1":-1},"topicB":{"0":-1}} latest

batch query

批量查询

批量查询结束时的结束点,可以是最新引用的“latest”,也可以是指定每个TopicPartition结束偏移量的json字符串。在json中,-1作为偏移量可以用于引用最新的,-2(最早的)作为偏移量是不允许的。
failOnDataLoss true or false true

streaming query

流式查询

当可能丢失数据时是否让查询失败(例如,主题被删除或偏移超出范围)。这可能是一个虚惊。当它不像你期望的那样工作时,你可以禁用它。如果由于丢失数据而无法从提供的偏移量中读取任何数据,批量查询将始终失败。
kafkaConsumer.pollTimeoutMs long 512 streaming and batch 在executors中轮询来自Kafka的数据的超时时间(以毫秒为单位)
fetchOffset.numRetries int 3 streaming and batch 在放弃提取Kafka偏移量之前重试的次数。
fetchOffset.retryIntervalMs long 10 streaming and batch 重试去提取Kafka偏移量之前等待的毫秒数。
maxOffsetsPerTrigger long none streaming and batch 每次触发间隔处理的最大偏移量的速率限制。指定的偏移量总数将按不同卷的topicPartition成比例地分割。

3. Writing Data to Kafka 将数据写入到Kafka

这里,我们描述了向Apache Kafka写入流式查询和批量查询的支持。注意Apache Kafka只支持至少一次写入语义。因此,在向Kafka写入流式查询或批量查询时,可能会复制一些记录;例如,如果Kafka需要重试未被Broker确认的消息,即使该Broker接收并编写了消息记录,也会发生这种情况。由于这些Kafka写入语义,结构化流不能阻止这种复制发生。然而,如果编写查询是成功的,那么你可以假定查询输出至少写入一次。当读取写入的数据删除重复项的可能解决方法可能是引入主要(唯一)key,当读取时这key可以用于执行重复数据删除。

写入到Kafka的Dataframe应该在模式上有以下列:

Column Type
key (optional) string or binary
value (required) string or binary
topic (*optional) string

* 如果“topic”配置选项不指定,则topic列是必须的。

value列是唯一要求的选项。如果key列没有指定,那么将会自动添加值为null的key列(查看Kafka语义中关于如何处理空值key)。如果topic列存在,那么在将给予的行写入到Kafka时,它的值用作topic,除非设置好“topic”配置选项。例如,“topic”配置选项覆盖topic列。

对于批量和流式查询,必须为Kafka sink设置以下选项:

Option value meaning
kafka.bootstrap.servers A comma-separated list of host:port Kafka "bootstrap.servers"配置。

 以下选项是可选的:

Option value default query type meaning
topic string none streaming and batch 设置所有行将在Kafka中写入的topic。该选项将覆盖数据中可能存在的任何topic列。

3.1 Creating a Kafka Sink for Streaming Queries 为流式查询创建Kafka Sink

Structured Streaming + Kafka Integration Guide 结构化流+Kafka集成指南 (Kafka broker version 0.10.0 or higher)

3.2 Writing the output of Batch Queries to Kafka 将批量查询的输出写入到Kafka中

Structured Streaming + Kafka Integration Guide 结构化流+Kafka集成指南 (Kafka broker version 0.10.0 or higher)

4. Kafka Specific Configurations Kafka特定的配置

Kafka自己的配置可以通过带有Kafka.prefix的DataStreamReader.option进行设置。例如,stream.option("kafka.bootstrap.servers","host":"port")。有关可能的Kafka参数,参阅Kafka消费者配置文档以获取与读取数据相关的参数,以及Kafka生产者配置文件以获取与写入数据相关的参数。

注意以下Kafka参数不能设置,Kafka source或者sink将会抛出错误。

  • group.id: Kafka source将会自动为每次查询创建唯一的分组id。
  • auto.offset.reset: 将source选项startingOffsets设置为指定从哪里开始。结构化流管理内部消费的偏移量,而不是依赖Kafka消费者来完成。这将确保在topic/partitons动态订阅时不会遗漏任何数据。注意,只有在启动新的流式查询时才会应用startingOffsets,并且恢复操作始终会从查询停止的位置启动。.
  • key.deserializer:Keys总是被反序列化为ByteArrayDeserializer的字节数组。使用DataFrame操作显式反序列化keys。
  • value.deserializer:Values总是被反序列化为ByteArrayDeserializer的字节数组。使用DataFrame操作显式反序列化values。
  • key.serializer: keys总是使用ByteArraySerializer或StringSerializer进行序列化。使用DataFrame操作将keys显示序列化为字符串或字节数组。
  • value.serializer: values总是使用ByteArraySerializer或StringSerializer进行序列化使用DataFrame操作将values显示序列化为字符串或字节数组。
  • enable.auto.commit: Kafka source不提交任何offset。
  • interceptor.classes: Kafka source总是以字节数组的形式读取key和value。使用ConsumerInterceptor是不安全的,因为它可能会打断查询。

5. Deploying 部署

与任何Spark应用程序一样,spark-submit用于启动你的应用程序。spark-sql-kafka-0-10_2.11及其依赖关系可以直接添加到使用--packages的spark-submit中,例如,

Structured Streaming + Kafka Integration Guide 结构化流+Kafka集成指南 (Kafka broker version 0.10.0 or higher)

更多关于提交带有外部依赖项的应用程序的详细信息参阅Application Submission Guide