Spark Structured Streaming流流聚合连接后无法写入ADLS求助

问题描述 #

平台：Databricks Notebooks | 语言：PySpark
背景：构建流数据管道节点，通过实时连接过滤符合行计数检查（count(*) == expected_count）的记录，但执行writeStream后仅创建ADLS文件夹，无Parquet数据写入。joined_df可正常写入，但joined_with_count_df及final_df无数据输出。

代码步骤：

# 1. 读取表为流DataFrame
table1_df = spark.readStream.format("delta").load(table1_path)
table2_df = spark.readStream.format("delta").load(table2_path).withColumn("insertion_time", current_timestamp())

# 2. 执行流连接
joined_df = (table1_df.join(table2_df, field == field, "inner")

# 3. 创建统计行数的DataFrame
count_df = joined_df.withWatermark("insertion_time", "1 second").groupBy("insertion_time", "Id").agg(count("*").alias("current_count"))

# 4. 将统计结果连接到连接后的DataFrame
joined_with_count_df = joined_df.join(count_df, on=field, how="inner").withColumn("row_count_check", "b.current_count" == "a.expected_count")

# 5. 过滤出行数匹配的记录
final_df = joined_with_count_df.filter("row_count_check" == True)

# 6. 写入流
final_df.writeStream.format("delta").outputMode("append").option("checkpointLocation", c_path).start(output_path)

核心问题与修复方案 #

你的代码存在多个逻辑与语法错误，导致无数据输出：

1. 流Join语法与条件错误 #

• 原代码joined_df缺少闭合括号，会直接报错；
• 关联条件field == field无效，必须指定具体关联列（如table1_df.Id == table2_df.Id），否则无法正确关联两个流表。

2. 字符串比较导致逻辑全错 #

• 创建row_count_check时，用字符串字面量"b.current_count" == "a.expected_count"做比较，会直接返回False（两个字符串不相等），而非比较列值；
• 过滤条件"row_count_check" == True是将字符串"row_count_check"与布尔值True比较，结果永远为False，导致无数据通过过滤。

3. Join未指定表别名 #

关联joined_df与count_df时引用了a.和b.别名，但未给DataFrame设置别名，导致列引用无效。

4. 聚合分组精度过高 #

insertion_time由current_timestamp()生成（精确到毫秒），按该字段分组后每个分组数据极少，甚至无数据，导致后续Join无法匹配。

修正后的完整代码 #

from pyspark.sql import functions as F

# 1. 读取流表
table1_df = spark.readStream.format("delta").load(table1_path)
table2_df = spark.readStream.format("delta").load(table2_path).withColumn("insertion_time", F.current_timestamp())

# 2. 修正流Join：指定有效关联列，补全语法
joined_df = table1_df.join(table2_df, table1_df["Id"] == table2_df["Id"], "inner")

# 3. 修正聚合：调整watermark窗口，截断时间避免分组过细
count_df = joined_df.withWatermark("insertion_time", "1 minute") \
    .groupBy(F.window("insertion_time", "1 minute"), "Id") \
    .agg(F.count("*").alias("current_count")) \
    .select(F.col("window.start").alias("window_start"), "Id", "current_count")

# 4. 修正关联与列计算：添加表别名，用列引用做比较
joined_with_count_df = joined_df.alias("a") \
    .join(count_df.alias("b"), on="Id", how="inner") \
    .withColumn("row_count_check", F.col("b.current_count") == F.col("a.expected_count"))

# 5. 修正过滤条件：直接引用布尔列
final_df = joined_with_count_df.filter(F.col("row_count_check"))

# 6. 写入流：确保checkpoint路径唯一且权限正常
query = final_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", c_path) \
    .start(output_path)

# 调试时可添加等待语句，查看实时输出
# query.awaitTermination()

额外调试建议 #

• 中间步骤添加控制台输出，验证数据是否符合预期：

  joined_with_count_df.writeStream.format("console").outputMode("append").start().awaitTermination()
  

• 检查expected_count字段是否存在于joined_df中，确认字段名拼写正确；
• 查看Databricks流查询日志（Jobs/Streams页面），排查权限或数据异常；
• 确保ADLS路径权限正确，集群拥有读写权限。

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