Spark Streaming forEachBatch写入Aurora时出现数据错位问题

问题出在foreachBatch中使用的lambda表达式对循环变量table的延迟绑定特性上：

• Python的闭包（比如这里的lambda）不会捕获循环变量的当前值，而是在lambda实际执行时才去查找变量的当前值。
• 你通过循环遍历tablesDF.items()创建各流的写入任务，当第一个批次触发执行lambda时，整个循环已经执行完毕，此时table变量的值是循环最后一次迭代的表名，或者因多线程并行执行的时序问题，拿到循环过程中某个不确定的table值。这就导致不同流的批量写入错误地使用同一个table参数，最终出现数据串表的情况。

核心是让每个lambda捕获当前循环迭代的table即时值，可通过以下两种方式实现：

方式1：用默认参数绑定当前值 #

修改foreachBatch的lambda定义，将table作为默认参数传入，每次循环迭代时都会把当前table值绑定到lambda参数上：

vars()[table+'_query'] = df.writeStream\
            .trigger(processingTime='120 seconds') \
            .foreachBatch(lambda fdf, batch_id, tbl=table: writeToAurora(fdf, batch_id, tbl)) \
            .option("checkpointLocation", f"s3://{bucket}/temporary/checkpoint/{table}")\
            .start()

方式2：用工厂函数生成lambda #

创建专门函数生成绑定当前table值的lambda，规避闭包延迟绑定问题：

def create_write_func(target_table):
    def write_func(fdf, batch_id):
        writeToAurora(fdf, batch_id, target_table)
    return write_func

# 循环中使用：
vars()[table+'_query'] = df.writeStream\
            .trigger(processingTime='120 seconds') \
            .foreachBatch(create_write_func(table)) \
            .option("checkpointLocation", f"s3://{bucket}/temporary/checkpoint/{table}")\
            .start()

额外优化建议 #

尽量避免vars()和eval()这类动态变量操作，改用字典存储各流的query对象，减少潜在问题：

query_dict = {}
for table, tableDF in tablesDF.items():
    df = tableDF.withColumn('csvData', F.from_csv('finalData', schema=tableSchema[table], options={'sep': '|','quote': '"'}))\
        .select('csvData.*')
    
    query_dict[table] = df.writeStream\
                .trigger(processingTime='120 seconds') \
                .foreachBatch(lambda fdf, batch_id, tbl=table: writeToAurora(fdf, batch_id, tbl)) \
                .option("checkpointLocation", f"s3://{bucket}/temporary/checkpoint/{table}")\
                .start()
                
for query in query_dict.values():
    query.awaitTermination()

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