python - Spark 2.0 DataFrame 初始化可能存在的错误

Question

以下代码可能会产生错误:

_struct = [
  types.StructField('string_field', types.StringType(), True),
  types.StructField('long_field', types.LongType(), True),
  types.StructField('double_field', types.DoubleType(), True)
]
_rdd = sc.parallelize([Row(string_field='1', long_field=1, double_field=1.1)])
_schema = types.StructType(_struct)
_df = sqlContext.createDataFrame(_rdd, schema=_schema)
_df.take(1)

预期的输出是应该创建一个包含 1 行的 RDD。

但是对于当前的行为，我收到以下错误:

DoubleType can not accept object '1' in type <type 'str'>

PS:我在 Scala 2.10 上使用 spark 2.0 编译

编辑

感谢回答者的建议，我现在可以正确理解这一点。为简化起见，请确保结构已排序。以下代码解释了这一点:

# This doesn't work:
_struct = [
  SparkTypes.StructField('string_field', SparkTypes.StringType(), True),
  SparkTypes.StructField('long_field', SparkTypes.LongType(), True),
  SparkTypes.StructField('double_field', SparkTypes.DoubleType(), True)
]
_rdd = sc.parallelize([Row(string_field='1', long_field=1, double_field=1.1)])

# But this will work, since schema is sorted:
_struct = sorted([
  SparkTypes.StructField('string_field', SparkTypes.StringType(), True),
  SparkTypes.StructField('long_field', SparkTypes.LongType(), True),
  SparkTypes.StructField('double_field', SparkTypes.DoubleType(), True)
], key=lambda x: x.name)
params = {'string_field':'1', 'long_field':1, 'double_field':1.1}
_rdd = sc.parallelize([Row(**params)])


_schema = SparkTypes.StructType(_struct)

_df = sqlContext.createDataFrame(_rdd, schema=_schema)
_df.take(1)

_schema = SparkTypes.StructType(_struct)

_df = sqlContext.createDataFrame(_rdd, schema=_schema)
_df.take(1)

Answer 1

这看起来像是 1.x 和 2.x 之间的行为变化，但我怀疑这是一个错误。特别是当您使用 kwargs(命名参数)创建 Row 对象时 the fields are sorted by names .让我们用一个简单的例子来说明:

Row(string_field='1', long_field=1, double_field=1.1)
## Row(double_field=1.1, long_field=1, string_field='1'

如您所见，字段顺序发生变化，不再反射(reflect)在架构中。

2.0.0 之前的 Spark 验证类型 only if data argument for createDataFrame is a local data structure .所以下面的代码:

sqlContext.createDataFrame(
    data=[Row(string_field='1', long_field=1, double_field=1.1)], 
    schema=_schema
)

在 1.6 中也会失败

推出 Spark 2.0.0 verification for RDDs并在本地和分布式输入之间提供一致的行为。