如何使用PySpark将数据保存为Oracle的VARCHAR2类型列？

Spark JDBC没有像Pandas to_sql那样直接的dtypes参数，但可以通过以下几种方式实现强制指定Oracle列类型，解决ROW_ID被存为CLOB的问题：

方法1：使用createTableColumnTypes参数直接指定建表类型 #

Spark JDBC提供了createTableColumnTypes选项，允许你指定创建表时各列的数据库原生类型。当写入模式为overwrite时，Spark会用这个定义来创建表，覆盖默认的类型映射：

delta_mlog_df = df.select(F.col("ROW_ID"), F.explode(F.col("TRANSACTIONS")).alias("TRANSACTION_ID"))
delta_mlog_df.write \
    .format("jdbc") \
    .option("driver", "oracle.jdbc.driver.OracleDriver") \
    .option("dbtable", ORIGIN_MLOG_HIST_TABLE) \
    .option("url", ORACLE_URL) \
    .option("user", ORACLE_USER) \
    .option("password", ORACLE_PASSWORD) \
    # 明确指定列的Oracle类型，ROW_ID设为VARCHAR2(100)（Oracle ROWID长度固定为18，100足够）
    .option("createTableColumnTypes", "TRANSACTION_ID NUMBER, ROW_ID VARCHAR2(100)") \
    .mode('overwrite') \
    .save()

方法2：手动创建表后写入（更可控） #

如果希望表结构完全由自己定义，先在Oracle中手动创建目标表，再通过Spark写入时使用truncate选项避免重建表：

1. 在Oracle中创建表 #

CREATE TABLE ORIGIN_MLOG_HIST_TABLE (
    TRANSACTION_ID NUMBER,
    ROW_ID VARCHAR2(100) -- 按需调整长度，比如VARCHAR2(20)更紧凑
);

2. 修改Spark写入代码 #

使用truncate选项清空表数据但保留结构，代替默认overwrite会重建表的行为：

delta_mlog_df = df.select(F.col("ROW_ID"), F.explode(F.col("TRANSACTIONS")).alias("TRANSACTION_ID"))
delta_mlog_df.write \
    .format("jdbc") \
    .option("driver", "oracle.jdbc.driver.OracleDriver") \
    .option("dbtable", ORIGIN_MLOG_HIST_TABLE) \
    .option("url", ORACLE_URL) \
    .option("user", ORACLE_USER) \
    .option("password", ORACLE_PASSWORD) \
    .option("truncate", "true") \
    .mode('overwrite') \
    .save()

方法3：通过自定义SQL语句写入 #

将DataFrame注册为临时视图，然后执行显式类型转换的INSERT语句：

delta_mlog_df.createOrReplaceTempView("temp_mlog")

# 构造带类型转换的INSERT语句
insert_sql = f"""
INSERT INTO {ORIGIN_MLOG_HIST_TABLE} (TRANSACTION_ID, ROW_ID)
SELECT TRANSACTION_ID, CAST(ROW_ID AS VARCHAR2(100)) FROM temp_mlog
"""

# 执行SQL写入
spark.sql(insert_sql)

这种方法适合需要更复杂数据转换逻辑的场景，同样需要提前创建好目标表。

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