将大型 DataFrame 从 PySpark 写入 Kafka 遇到超时

Question

我正在尝试将包含大约 2.3 亿条记录的数据帧写入 Kafka。更具体地说是Kafka-enable Azure Event Hub，但我不确定这是否真的是我的问题的根源。

EH_SASL = 'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://myeventhub.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=****";'

dfKafka \
.write  \
.format("kafka") \
.option("kafka.sasl.mechanism", "PLAIN") \
.option("kafka.security.protocol", "SASL_SSL") \
.option("kafka.sasl.jaas.config", EH_SASL) \
.option("kafka.bootstrap.servers", "myeventhub.servicebus.windows.net:9093") \
.option("topic", "mytopic") \
.option("checkpointLocation", "/mnt/telemetry/cp.txt") \
.save()

这可以很好地启动并成功（并且非常快）将大约 3-4 百万条记录写入队列。但是几分钟后工作就停止了，并显示如下消息：

org.apache.spark.SparkException：作业因阶段故障而中止：阶段 7.0 中的任务 6 失败 4 次，最近一次失败：阶段 7.0 中丢失任务 6.3（TID 248、10.139.64.5、执行程序 1）：kafkashaded .org.apache.kafka.common.errors.TimeoutException：mytopic-18 的 61 条记录到期：自上次追加以来已过去 32839 毫秒

或

org.apache.spark.SparkException：作业因阶段故障而中止：阶段 8.0 中的任务 13 失败 4 次，最近一次失败：阶段 8.0 中丢失任务 13.3（TID 348、10.139.64.5、执行程序 1）：kafkashaded .org.apache.kafka.common.errors.TimeoutException: 请求超时。

另外，我从来没有看到正在创建/写入检查点文件。 我也玩过.option("kafka.delivery.timeout.ms", 30000) 和不同的值，但这似乎没有任何效果。

我在 Azure Databricks 集群 5.0 版（包括 Apache Spark 2.4.0、Scala 2.11）中运行它

我没有在我的 Event Hub 上看到任何错误，例如限制，所以应该没问题。

Answer 1

终于想通了（大部分）：

原来大约 16000 条消息的默认批量大小对于端点来说太大了。在我将 batch.size 参数设置为 5000 后，它工作并以每分钟大约 700k 条消息写入事件中心。此外，上面的超时参数是错误的，只是被忽略了。是kafka.request.timeout.ms

唯一的问题是它仍然随机运行超时并且显然又从头开始，所以我最终得到了重复。将为此打开another question。

dfKafka \
.write  \
.format("kafka") \
.option("kafka.sasl.mechanism", "PLAIN") \
.option("kafka.security.protocol", "SASL_SSL") \
.option("kafka.sasl.jaas.config", EH_SASL) \
.option("kafka.batch.size", 5000) \
.option("kafka.bootstrap.servers", "myeventhub.servicebus.windows.net:9093") \
.option("kafka.request.timeout.ms", 120000) \
.option("topic", "raw") \
.option("checkpointLocation", "/mnt/telemetry/cp.txt") \
.save()