如何将MySQL存储过程执行结果转换为PySpark DataFrame？

执行MySQL存储过程并尝试将结果转换为PySpark DataFrame，使用代码如下：

driver_manager = spark._sc._gateway.jvm.java.sql.DriverManager
connection = driver_manager.getConnection(args["sql_server_jdbc_url"], database_username, database_user_password)

exec_statement = connection.prepareCall("EXEC SP")
exec_statement.execute()
result = exec_statement.getResultSet()
from pyspark.sql import SQLContext, DataFrame
sqlContext = SQLContext(sparkContext=spark.sparkContext, sparkSession=spark)
df = DataFrame(result, sqlContext)
df.printSchema()

调用df.printSchema()时失败，抛出错误：Py4JError: An error occurred while calling o83.schema. Trace:

• 核心问题原因：PySpark的DataFrame构造函数无法直接接收Java的ResultSet对象，跨JVM对象直接交互会触发Py4J调用限制，导致报错。

• 方案一：Spark原生JDBC调用存储过程
  利用Spark内置的JDBC支持，让Spark自行处理连接和结果集转换，示例代码：

  # MySQL驱动为com.mysql.cj.jdbc.Driver，SQL Server驱动为com.microsoft.sqlserver.jdbc.SQLServerDriver
  df = spark.read \
      .format("jdbc") \
      .option("url", args["sql_server_jdbc_url"]) \
      .option("dbtable", "(EXEC SP) AS temp") \
      .option("user", database_username) \
      .option("password", database_user_password) \
      .option("driver", "com.mysql.cj.jdbc.Driver") \
      .load()
  
  df.printSchema()
  df.show()
  

  这种方式无需手动处理Java对象，Spark会自动完成结果集到DataFrame的转换，同时支持连接池等优化。

• 方案二：手动映射ResultSet构造DataFrame
  若必须手动管理JDBC连接，需自行遍历ResultSet提取元数据和数据，再构建DataFrame：

  from pyspark.sql.types import StructType, StructField, StringType, IntegerType
  import pyspark.sql.Row
  
  # 获取结果集元数据，构建Schema
  meta = result.getMetaData()
  column_count = meta.getColumnCount()
  schema_fields = []
  for i in range(1, column_count + 1):
      col_name = meta.getColumnName(i)
      # 根据实际字段类型调整映射，此处以StringType为例
      col_type = StringType()
      schema_fields.append(StructField(col_name, col_type, nullable=True))
  schema = StructType(schema_fields)
  
  # 遍历结果集生成Row列表
  rows = []
  while result.next():
      row_data = [result.getObject(i) for i in range(1, column_count + 1)]
      rows.append(pyspark.sql.Row(*row_data))
  
  # 创建DataFrame
  df = spark.createDataFrame(rows, schema)
  df.printSchema()
  
  # 释放资源
  result.close()
  exec_statement.close()
  connection.close()
  

  注意：需根据数据库字段的JDBC类型，手动映射为对应的PySpark数据类型；使用完毕后务必关闭JDBC相关资源，避免连接泄漏。

• 额外注意事项

  • 确保Spark环境已添加对应数据库的JDBC驱动包，可通过--jars参数提交任务或在集群环境中配置驱动依赖。
  • 若存储过程返回多个结果集，Spark JDBC方式仅会处理第一个结果集，多结果集场景建议使用手动遍历方案，通过exec_statement.getMoreResults()切换结果集。

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