使用Stream Analytics从ADLS Gen2的JSON文件向Event Hub发送事件

1. 验证JSON文件格式（优先级最高） #

Stream Analytics要求读取的JSON文件为JSON Lines格式（每行一个独立的JSON对象），如果Spark输出的是包裹在数组中的JSON（例如[{...}, {...}]），SA会将其识别为单个事件甚至无法解析，直接导致无输出。

• 打开Spark输出的JSON文件，确认是否每行一个独立JSON对象，而非数组结构。
• 如果Spark输出的是数组，修改Spark代码：使用toJSON().saveAsTextFile()直接输出每行JSON，或在write.json()时添加option("lineSep", "\n")参数强制按行输出。

2. 检查Stream Analytics输入配置细节 #

• 起始时间设置：若启动SA作业前已有文件存在，需将输入起始时间设为「自定义时间」并选择文件创建时间之前的时间，否则SA只会处理作业启动后新增的文件。
• 递归搜索与路径模式：如果文件存储在子目录中，确保输入配置开启「递归搜索」；同时检查路径模式是否匹配文件实际存储路径（例如*.json是否覆盖目标文件）。
• 序列化与压缩：确认输入配置中选择的格式为JSON，且未开启压缩（若Spark输出的文件未压缩）。

3. 排查水印延迟与超时问题 #

• 开启诊断日志：在Azure门户的SA作业中开启诊断日志（发送至Log Analytics），通过查询AzureDiagnostics表过滤OperationName为ProcessInputEvents或ProcessOutputEvents的记录，定位具体错误（如JSON解析失败、文件读取超时等）。
• 拆分大文件：若Spark输出单个超大文件（如GB级），SA处理时易超时。修改Spark代码，用repartition(n)将数据拆分为多个小文件（建议单文件大小控制在100MB以内），提升SA处理效率。
• 检查Event Hub吞吐量：查看Event Hub的「传入消息速率」「限流错误」指标，若吞吐量单位（TU）不足，会导致SA输出被限流，进而引发水印延迟升高。需根据实际流量调整TU数量。

4. 优化SA查询逻辑 #

SA的流处理依赖时间水印，若未明确指定时间戳列，会默认使用文件创建时间作为事件时间，若文件创建时间异常（如未来时间），会导致水印计算错误。

• 修改查询，指定JSON中的时间字段作为事件时间（若存在）：

  SELECT * INTO eventhub FROM JsonFiles TIMESTAMP BY event_time
  

• 若JSON无时间字段，可使用文件创建时间：

  SELECT * INTO eventhub FROM JsonFiles TIMESTAMP BY CreationTime
  

5. 验证文件处理状态 #

• SA处理ADLS文件时，会在存储账户的$logs目录生成处理记录（需提前开启存储日志），可通过这些日志查看文件是否被成功处理。
• 若之前有失败的文件记录，重启SA作业时选择「重置状态」，让SA重新处理所有符合条件的文件。

6. 最小化测试验证 #

手动创建一个符合JSON Lines格式的小文件（例如2-3行独立JSON），上传至ADLS目标路径，启动SA作业测试是否能正常输出到Event Hub，以此排除Spark输出文件的格式问题。

内容的提问来源于stack exchange，提问作者Bill Kelly