从Pyspark向PostgreSQL复合列表插入数据报错，如何解决？

Spark的STRUCT类型无法直接映射PostgreSQL的自定义复合类型（info_type），JDBC驱动无法自动完成这种类型转换，因此抛出类型不匹配错误。

方法1：用UDF转换Struct为PostgreSQL复合类型格式 #

定义UDF将Spark Struct对象转为PostgreSQL能识别的ROW(age, profession)格式字符串，再写入表中：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def struct_to_pg_complex(struct_obj):
    return f"ROW({struct_obj['age']}, '{struct_obj['profession']}')"

struct_udf = udf(struct_to_pg_complex, StringType())

# 转换info列格式
df_transformed = df.withColumn("info", struct_udf(df["info"]))

# 写入PostgreSQL
df_transformed.write.jdbc(
    postgres_db_url,
    "public.your_table",
    mode="append",
    properties=postgres_db_properties
)

方法2：用Spark内置函数拼接复合类型字符串 #

无需UDF，直接用concat函数拼接出符合PostgreSQL要求的格式：

from pyspark.sql.functions import concat, lit, col

df_transformed = df.withColumn(
    "info",
    concat(
        lit("ROW("),
        col("info.age"),
        lit(", '"),
        col("info.profession"),
        lit("')")
    )
)

# 写入操作
df_transformed.write.jdbc(
    postgres_db_url,
    "public.your_table",
    mode="append",
    properties=postgres_db_properties
)

注意事项 #

• 确保PostgreSQL JDBC驱动版本与Spark、PostgreSQL版本兼容，推荐使用postgresql-42.6.0及以上版本。
• 如果profession字段包含单引号等特殊字符，需要添加转义逻辑，避免SQL语法错误。

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