Spark-Kusto Connector写入流仅处理1批后闲置及高延迟问题咨询

一、当前写入延迟/闲置问题的核心原因 #

1. Connector版本的流式支持局限 #

你使用的com.microsoft.azure.kusto:kusto-spark_3.0_2.12:5.0.4版本无原生流式写入能力，当前的writeStream是基于Spark微批+ADX队列摄入（Queued模式）的模拟实现，并非端到端的真正流式摄入：

• Queued模式依赖ADX后台队列的批处理逻辑，即便ADX侧配置了低延迟，Connector端仍会按Spark微批节奏攒数据后批量发送，无法实现毫秒级触发。
• 若Spark微批未触发新批次（默认触发逻辑依赖数据量/内部调度），就会出现处理完一批后进入闲置的情况。

2. Spark微批触发配置缺失 #

你的代码未显式设置Spark Structured Stream的trigger参数：

• 默认情况下，Spark采用ProcessingTime触发，但无明确间隔时，触发逻辑依赖数据到达量与内部调度，导致触发不规律，出现数分钟延迟或闲置。

3. 参数配置不匹配低延迟场景 #

• clientBatchingLimit="100"是Queued模式下的客户端批量阈值，若Spark微批输出记录数不足100，Connector会等待攒够数量再发送，进一步加剧延迟。
• 未针对低延迟场景配置Spark侧的流处理参数（如微批间隔、最小触发阈值等）。

二、关于PR#301的Stream写入模式问题 #

• 发布状态：PR#301已合并至Connector主分支，对应的原生流式写入模式（writeMode="Stream"）在5.1.0及以上版本以Beta形式发布，可用于测试。
• 时间线：该功能在2023年下半年完成合并，5.1.0版本正式提供Beta支持，后续版本逐步优化稳定性，当前最新稳定版已将其纳入正式功能范畴。

三、实现毫秒级低延迟的解决方案 #

1. 升级Connector版本并启用原生流式模式 #

将Maven依赖升级至5.1.0及以上版本（推荐最新稳定版），替换为原生Stream模式：

options = {
    "kustoCluster": f"{kusto_cluster}",
    "kustoDatabase": f"{kusto_db}",
    "kustoTable": f"{table}",
    "kustoAadAppId": f"{KUSTO_AAD_APP_ID}",
    "kustoAadAppSecret": f"{KUSTO_AAD_APP_SECRET}",
    "kustoAadAuthorityID": f"{KUSTO_AAD_AUTHORITY_ID}",
    "writeMode" : "Stream",  # 启用原生流式写入
    "streamingIngestionBatchSize": "1000",  # 单批次最大记录数，按需调整
    "streamingIngestionBatchTimeoutMs": "500"  # 超时时间，到点即使未达批次大小也发送
}

kust_stream = (df
                .writeStream
                .queryName("ADX_WRITE")
                .format("com.microsoft.kusto.spark.datasink.KustoSinkProvider")
                .options(**options)
                .trigger(processingTime='500 milliseconds')  # 设置微批触发间隔
                .start()
)

kust_stream.awaitTermination()

2. 配置Spark低延迟参数 #

在Databricks集群配置中添加以下参数，优化微批触发效率：

spark.sql.streaming.minBatchesToRetain=1
spark.streaming.backpressure.enabled=true  # 启用背压适配数据波动
spark.sql.streaming.kafkaConsumer.pollTimeoutMs=100  # 数据源为Kafka时按需调整

3. 验证ADX侧低延迟配置 #

确保ADX表的流式摄入策略已启用，且数据库IngestionPolicy中IngestionBatchingPolicy的MaximumBatchingTimeSpan设为00:00:01级别的低延迟阈值。

内容的提问来源于stack exchange，提问作者Raghu Sivakumar