带有架构的行中的 sqlContext.createDataframe。 pyspark: TypeError: IntegerType can't accept object in type <type 'unicode'>答案

【问题标题】：sqlContext.createDataframe from Row with Schema. pyspark: TypeError: IntegerType can not accept object in type <type 'unicode'>带有架构的行中的 sqlContext.createDataframe。 pyspark: TypeError: IntegerType can't accept object in type <type 'unicode'>
【发布时间】：2015-11-03 13:16:36
【问题描述】：

花了很多时间弄清楚为什么会出现以下错误

pyspark: TypeError: IntegerType can not accept object in type <type 'unicode'>

在尝试基于行和架构创建数据框时，我注意到以下几点：

在我的 rdd 中有一个名为 rrdRows 的 Row，如下所示：

Row(a="1", b="2", c=3)

我的 dfSchema 定义为：

dfSchema = StructType([
        StructField("c", IntegerType(), True),
        StructField("a", StringType(), True),
        StructField("b", StringType(), True)
        ])

如下创建数据框：

df = sqlContext.createDataFrame(rddRows, dfSchema)

带来了上面提到的Error，因为Spark只考虑了StructFields在schema中的顺序，并没有把StructFields的名字和Row字段的名字匹配起来。

换句话说，在上面的例子中，我注意到 spark 试图创建一个如下所示的数据帧（如果不存在 typeError。例如，如果所有内容都是 String 类型）

+---+---+---+
| c | b | a |
+---+---+---+
| 1 | 2 | 3 |
+---+---+---+

这真的是预期的，还是某种错误？

编辑：rddRows 是按照这些思路创建的：

def createRows(dic):
    res = Row(a=dic["a"],b=dic["b"],c=int(dic["c"])
    return res

rddRows = rddDict.map(createRows)

其中 rddDict 是已解析的 JSON 文件。

【问题讨论】：

你是如何创建你的rddRows的？
代码对于注释来说有点大，但我按照这些思路做： def createRows(dic): res = Row(a=dic["a"],b=dic[" b"],c=int(dic["c"]) return res rddRows = rddDict.map(createRows) 其中 rddDict 是解析的 JSON 文件。但是，我也尝试了另一个示例，但得到了相同的结果。
类型是：。我在火花流中使用它，但我也在一个非常简单的批处理作业中观察到了同样的问题。
嗯，这看起来像是预期的行为。 PySpark Row，类似于它的 Scala 对应物，只是一个元组。这意味着它具有固定的值和大小顺序。其他所有内容，例如名称或模式（如果是 Scala 版本），都只是元数据。由于行可能根本没有名称，或者模式中的名称可能与行中的名称不同，因此唯一合理的匹配是顺序。这与例如 JSON 源相反，其中顺序没有意义，名称是匹配记录的唯一好方法。
嗯好的。感谢您的澄清。也许作为一个简短的跟进：假设我已经有另一个包含 c、b、a 列的数据帧，我想将其附加到上面创建的数据帧中。实现它的最佳方式是什么？我想到了 .unionAll 函数。但是，为了使用它，我需要两个数据框的列顺序相同，对吧？

标签： unicode apache-spark apache-spark-sql pyspark spark-dataframe

【解决方案1】：

如果您提供关键字参数，Row 的构造函数会对键进行排序。看看源代码here。当我发现这一点时，我最终对我的schema 进行了相应的排序，然后再将其应用于数据框：

   sorted_fields = sorted(dfSchema.fields, key=lambda x: x.name)
   sorted_schema = StructType(fields=sorted_fields)
   df = sqlContext.createDataFrame(rddRows, sorted_schema)

【讨论】：