尝试使用SparkSQL分解“B列”数组

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

spark = SparkSession.builder.appName("Test_app").getOrCreate()

input_data = [
    (1, [{"id":"123","phone":"124"}]),
    (3, [{"id":"456","phone":"741"}])
]

schema = StructType([
    StructField("Column A", IntegerType(), True),
    StructField("Column B", ArrayType(StructType([
        StructField("id", StringType(), True),
        StructField("phone", StringType(), True)
    ])), True)
])

df = spark.createDataFrame(input_data, schema)

df_exploded = df.selectExpr("Column A", "explode(Column B) as e") \
                .select("e.id", "e.phone")

df_exploded.show()

输出低于;

+---+-----+
| id|phone|
+---+-----+
|123|  124|
|456|  741|
+---+-----+

将其转换为一个rdd，然后读取为json。为了测试，我删除了第二行中的id元素。

input_data = [
    (1, [{"id":"123","phone":"124"}]),
    (3, [{"phone":"741"}])
]
df = spark.createDataFrame(input_data, ["ColA","ColB"])
spark.read.json(df.rdd.map(lambda r: r.ColB)).show()

+----+-----+
|  id|phone|
+----+-----+
| 123|  124|
|null|  741|
+----+-----+