如何将PostgreSQL的JSON数组列转换为Spark DataFrame？

你解析返回null的核心原因是：目标列存储的是JSON数组，但你直接用StructType定义schema，正确的schema需要用ArrayType包裹StructType，匹配数组的结构。以下是完整的解决步骤和代码示例：

1. 定义匹配JSON数组的Schema #

针对每个数组对象包含numero、resposta、peso字段的结构，定义数组类型的schema：

Scala版本 #

import org.apache.spark.sql.types._

val jsonSchema = ArrayType(StructType(Seq(
  StructField("numero", IntegerType, nullable = true),
  StructField("resposta", StringType, nullable = true),
  StructField("peso", DoubleType, nullable = true)
)))

Python版本 #

from pyspark.sql.types import ArrayType, StructType, StructField, IntegerType, StringType, DoubleType

json_schema = ArrayType(StructType([
    StructField("numero", IntegerType(), nullable=True),
    StructField("resposta", StringType(), nullable=True),
    StructField("peso", DoubleType(), nullable=True)
]))

2. 读取PostgreSQL数据 #

JDBC读取PostgreSQL的jsonb列时，Spark会将其识别为字符串类型，直接读取即可：

Scala版本 #

val df = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://your-host:5432/your-db")
  .option("dbtable", "your-target-table")
  .option("user", "your-username")
  .option("password", "your-password")
  .option("driver", "org.postgresql.Driver")
  .load()

Python版本 #

df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://your-host:5432/your-db") \
    .option("dbtable", "your-target-table") \
    .option("user", "your-username") \
    .option("password", "your-password") \
    .option("driver", "org.postgresql.Driver") \
    .load()

3. 解析JSON数组列 #

使用from_json函数，传入正确的数组schema解析目标列：

Scala版本 #

import org.apache.spark.sql.functions._

val parsedDF = df.withColumn("parsed_json_array", from_json(col("your-jsonb-column-name"), jsonSchema))

Python版本 #

from pyspark.sql.functions import from_json, explode

parsed_df = df.withColumn("parsed_json_array", from_json(df["your-jsonb-column-name"], json_schema))

4. 展开数组为结构化行（可选） #

如果需要将数组中的每个对象拆分为单独的行，使用explode函数：

Scala版本 #

val finalDF = parsedDF
  .select("*", explode(col("parsed_json_array")).alias("json_object"))
  .select(
    // 保留原表其他字段，按需调整
    col("id"),
    col("json_object.numero").alias("numero"),
    col("json_object.resposta").alias("resposta"),
    col("json_object.peso").alias("peso")
  )

Python版本 #

final_df = parsed_df.select("*", explode("parsed_json_array").alias("json_object")) \
    .select(
        "id",
        "json_object.numero",
        "json_object.resposta",
        "json_object.peso"
    )

最终效果 #

处理后得到的结构化DataFrame结构如下（以示例数据为例）：

排查小贴士 #

• 如果解析仍返回null：检查JSON字符串是否格式合法（比如是否有未转义的特殊字符），确认schema的字段类型与JSON中实际数据类型完全匹配（比如numero是数字还是字符串）
• 如需保留数组结构：直接使用parsedDF，可通过parsed_json_array[index].field的方式访问数组内元素

内容的提问来源于stack exchange，提问作者Vitor L.