创建Spark DataFrame时遇TypeError: Unable to infer the type of the field _jdf问题的解决咨询

问题原因 #

你遇到的这个错误，核心原因是你往列表a中添加的是一个个Spark DataFrame对象，但spark.createDataFrame()的输入需要是行数据集合（比如Row对象列表、Python字典列表、RDD等），而不是DataFrame的列表。Spark无法推断DataFrame对象的结构类型，所以抛出了这个类型错误。

你的循环逻辑里，json_df.select(...)返回的是一个完整的Spark DataFrame，而不是单条数据，把这些DataFrame堆进列表再传给createDataFrame，自然会出问题。

推荐解决方案：用Spark内置函数展开嵌套数组（高效且符合Spark设计） #

你的需求是展开多层嵌套的数组（cc、at、dapi都是数组类型），Spark提供了explode函数专门处理这种场景，完全不需要手动循环索引，而且能利用Spark的分布式计算能力，效率远超手动循环。

具体代码如下：

from pyspark.sql.functions import explode

# 第一步：展开cc数组，把每个cc元素拆成单独的行
df_step1 = json_df.withColumn("cc_exploded", explode(json_df["c1"]["l1"]["cc"]))

# 第二步：基于上一步的结果，展开at数组
df_step2 = df_step1.withColumn("at_exploded", explode(df_step1["cc_exploded"]["at"]))

# 第三步：展开dapi数组，得到所有嵌套元素的笛卡尔积
df_step3 = df_step2.withColumn("dapi_exploded", explode(df_step2["at_exploded"]["dapi"]))

# 最后选择需要的字段并完成重命名
df_cart_rec = df_step3.select(
    df_step3["cc_exploded"]["nm"].alias("cl_nm"),
    df_step3["cc_exploded"]["perc"].alias("cl_perc"),
    df_step3["at_exploded"]["link1"].alias("cl_at_link1"),
    df_step3["at_exploded"]["link2"].alias("cl_at_link2"),
    df_step3["at_exploded"]["lg"].alias("cl_at_lg"),
    df_step3["at_exploded"]["nm"].alias("cl_at_nm"),
    df_step3["at_exploded"]["perc"].alias("cl_at_perc"),
    df_step3["dapi_exploded"]["cp"].alias("cl_at_dapi_cp"),
    df_step3["dapi_exploded"]["nm"].alias("cl_at_dapi_nm"),
    df_step3["dapi_exploded"]["vl"].alias("cl_at_dapi_vl")
)

这个方案的优势：

• 完全避免了Driver端的数据拉取，不会出现内存溢出问题
• 自动处理所有数组元素，不需要手动计算num_i、num_j、num_k的长度
• 符合Spark的分布式计算模型，处理大数据量时性能远优于手动循环

备选方案：修改循环逻辑（不推荐，仅适用于极小数据集） #

如果你坚持要用循环的方式（不建议，大数据场景下会出问题），需要把每个select后的DataFrame中的行数据提取出来，转换成Spark能识别的格式（比如Row对象）再存入列表：

a = []
if num_i != 0:
    for i in range(num_i-1):
        for j in range(num_j-1):
            for k in range(num_k-1):
                # 获取当前索引对应的字段组成的临时DataFrame
                temp_df = json_df.select(
                    json_df["c1"]["l1"]["cc"][i]["nm"].alias("cl_nm"),
                    json_df["c1"]["l1"]["cc"][i]["perc"].alias("cl_perc"),
                    json_df["c1"]["l1"]["cc"][i]["at"][j]["link1"].alias("cl_at_link1"),
                    json_df["c1"]["l1"]["cc"][i]["at"][j]["link2"].alias("cl_at_link2"),
                    json_df["c1"]["l1"]["cc"][i]["at"][j]["lg"].alias("cl_at_lg"),
                    json_df["c1"]["l1"]["cc"][i]["at"][j]["nm"].alias("cl_at_nm"),
                    json_df["c1"]["l1"]["cc"][i]["at"][j]["perc"].alias("cl_at_perc"),
                    json_df["c1"]["l1"]["cc"][i]["at"][j]["dapi"][k]["cp"].alias("cl_at_dapi_cp"),
                    json_df["c1"]["l1"]["cc"][i]["at"][j]["dapi"][k]["nm"].alias("cl_at_dapi_nm"),
                    json_df["c1"]["l1"]["cc"][i]["at"][j]["dapi"][k]["vl"].alias("cl_at_dapi_vl")
                )
                # 把临时DataFrame的所有行提取出来，添加到列表a中
                a.extend(temp_df.collect())

# 用Row对象列表创建最终的DataFrame
df_cart_rec = spark.createDataFrame(a)

⚠️ 注意：collect()会把分布式存储的数据全部拉取到Driver节点的内存中，如果你的数据集很大，很容易出现OOM（内存溢出）错误，所以强烈推荐第一种方案。

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