如何使用PySpark向PostgreSQL的jsonb列插入数据？

核心问题分析 #

报错本质是Spark传递的stores列是字符串类型（character varying），但PostgreSQL目标列是jsonb类型，两者类型不匹配，必须显式完成类型转换。

解决方案1：直接利用PostgreSQL类型转换（最简单） #

CSV中的stores列本身就是合法的JSON数组字符串（如[28,29,...]），无需额外拆分，只需在写入时让PostgreSQL自动转换类型：

from pyspark.sql.functions import col

df = spark.read \
    .option("header", True) \
    .format("csv") \
    .load(path)

# 通过子查询实现字符串到jsonb的转换
df.select(col("stores")) \
    .write \
    .format("jdbc") \
    .option("url", db_url) \
    .option("driver", "org.postgresql.Driver") \
    .option("dbtable", f"(SELECT CAST(stores AS jsonb) AS stores FROM {target_table}) AS temp_insert") \
    .option("user", db_user) \
    .option("password", db_password) \
    .mode("append") \
    .save()

解决方案2：Spark内解析为数组再转JSON写入 #

如果担心CSV中stores格式不规范（比如存在多余空格），可以先解析为整数数组，再转为标准JSON字符串后写入：

from pyspark.sql.functions import regexp_replace, split, cast, to_json, col

df = spark.read \
    .option("header", True) \
    .format("csv") \
    .load(path) \
    # 去除括号和空格，分割为字符串数组并转为整数数组
    .withColumn("stores_array", split(regexp_replace(col("stores"), r'^\[|\]|\s+', ''), ",").cast("array<int>")) \
    # 将整数数组转为标准JSON字符串
    .withColumn("stores", to_json(col("stores_array"))) \
    .drop("stores_array")

# 写入时指定字符串类型为JSON格式
df.select(col("stores")) \
    .write \
    .format("jdbc") \
    .option("url", db_url) \
    .option("driver", "org.postgresql.Driver") \
    .option("dbtable", target_table) \
    .option("user", db_user) \
    .option("password", db_password) \
    .option("stringtype", "unspecified") \
    .mode("append") \
    .save()

解决方案3：调用PostgreSQL的to_jsonb函数转换 #

直接在写入的SQL语句中调用PostgreSQL内置函数完成转换：

df.select(col("stores")) \
    .write \
    .format("jdbc") \
    .option("url", db_url) \
    .option("driver", "org.postgresql.Driver") \
    .option("dbtable", f"INSERT INTO {target_table} (stores) VALUES (to_jsonb(?))") \
    .option("user", db_user) \
    .option("password", db_password) \
    .mode("append") \
    .save()

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