在AWS Glue Spark中从DataFrame提取SQL查询并执行获取结果

核心思路 #

从source_df中提取存储SQL语句的目标列，将SQL语句收集到Driver端后逐个执行JDBC查询，最终合并所有查询结果得到final_df2。

具体实现（代码示例） #

假设存储SQL语句的列名为sql_statement，请替换为你实际的列名：

1. 提取并收集SQL语句
   将DataFrame中的SQL语句收集到本地列表（注：若source_df数据量极大，建议分批处理，避免Driver内存溢出）：

   # 提取目标列并转换为本地列表
   sql_queries = source_df.select("sql_statement").rdd.flatMap(lambda x: x).collect()
   

2. 遍历执行SQL并合并结果
   初始化空DataFrame，逐个执行收集到的SQL，将结果合并：

   from pyspark.sql import DataFrame
   
   # 初始化最终结果DataFrame
   final_df2 = None
   
   for query in sql_queries:
       # 跳过空SQL语句
       if not query.strip():
           continue
       try:
           # 执行当前SQL查询
           temp_df = spark.read.format("jdbc") \
               .option("url", Oracle_jdbc_url) \
               .option("query", query) \
               .option("user", Oracle_Username) \
               .option("password", Oracle_Password) \
               .load()
           # 合并结果到final_df2
           if final_df2 is None:
               final_df2 = temp_df
           else:
               # 兼容Schema不一致场景，可根据业务调整
               final_df2 = final_df2.unionByName(temp_df, allowMissingColumns=True)
       except Exception as e:
           # 捕获异常避免流程中断，记录错误信息
           print(f"执行SQL失败: {query}，错误详情: {str(e)}")
   

关键注意事项 #

• Schema兼容性：若不同SQL的查询结果结构不一致，unionByName配合allowMissingColumns=True可兼容字段缺失场景，但需确保符合业务逻辑。
• 性能优化：若SQL语句数量过多，建议改用foreachPartition在Executor端分批执行，减少Driver端内存压力。
• 安全风险：需确保table1中存储的SQL语句为可信来源，避免SQL注入风险。

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