如何在PySpark中将SQL cursor输出转换为Spark DataFrame并生成parquet文件

你通过cursor.fetchall()拿到的是行数据组成的元组列表，首先需要同步获取查询的列名，才能正确构造Spark DataFrame的Schema，具体实现步骤如下：

步骤1：获取原始数据与列名 #

首先从游标中同时取出数据和字段信息：

# 假设你已经初始化了数据库连接和游标cursor，且已执行完查询语句
raw_data = cursor.fetchall()
# 从cursor.description中提取列名，最终返回所有字段的名称列表
columns = [desc[0] for desc in cursor.description]

步骤2：构造Spark DataFrame #

根据数据量大小，可选择两种不同的处理方案：

• 小数据量场景（raw_data可完全放入内存）：直接用createDataFrame方法构造

from pyspark.sql import SparkSession

# 初始化SparkSession，如果你已经初始化过可以跳过这步
spark = SparkSession.builder.appName("CursorToDF").getOrCreate()

# 直接转换，传入数据和列名，Spark会自动推断字段类型
df = spark.createDataFrame(raw_data, schema=columns)

# 如果需要精确控制字段类型、避免自动推断出错，可以手动定义Schema
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType

# 示例Schema，需要根据你实际的字段数量、类型做调整
custom_schema = StructType([
    StructField("id", IntegerType(), nullable=True),
    StructField("name", StringType(), nullable=False),
    StructField("amount", DoubleType(), nullable=True)
])
df = spark.createDataFrame(raw_data, schema=custom_schema)

• 大数据量场景：如果fetchall()返回的数据量过大会撑爆Driver节点内存，建议直接用Spark自带的JDBC数据源连接数据库读取，不需要经过Python cursor中转，性能和稳定性更高：

df = spark.read.format("jdbc") \
    .option("url", "jdbc:数据库类型://地址:端口/库名") \
    .option("dbtable", "你的查询语句或者表名") \
    .option("user", "数据库账号") \
    .option("password", "数据库密码") \
    .option("driver", "对应数据库的JDBC驱动类名") \
    .load()

步骤3：输出为Parquet文件 #

构造完DataFrame后直接调用write.parquet方法即可：

# 输出到指定路径，mode参数可选：overwrite覆盖/append追加/ignore忽略已存在的文件/error默认存在就报错
df.write.parquet("你要保存的parquet文件路径", mode="overwrite")

# 如果需要输出为单个parquet文件，可以先合并分区再输出
df.coalesce(1).write.parquet("保存路径", mode="overwrite")

注意：用Python cursor中转数据的方式仅适合小批量数据，大数据量下JDBC直读会自动分区拉取，不会把所有数据都加载到Driver节点，更适合生产环境使用。

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