用于Kafka 0.10的结构化流集成从Kafka读取数据并将数据写入到Kafka。
1. Linking
对于使用SBT/Maven项目定义的Scala/Java应用程序，用以下工件artifact连接你的应用程序：
对于Python应用程序，你需要在部署应用程序时添加上面的库及其依赖关系。查看Deploying子节点。
2. Reading Data from Kafka 从Kafka读取数据
2.1 Creating a Kafka Source for Streaming Queries 为流式查询创建一个Kafka来源
2.2 Creating a Kafka Source for Batch Queries 为批处理查询创建一个Kafka来源
如果你有一个更适合用于批处理的用例，你可以为定义的偏移量范围创建一个Dataset/DataFrame。
来源的每一行有以下格式：
| 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
3.2 Writing the output of Batch Queries to Kafka 将批量查询的输出写入到Kafka中
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中，例如，
更多关于提交带有外部依赖项的应用程序的详细信息参阅Application Submission Guide。