python之正确读取 PySpark 文件中的类型_编程设计

我有一个包含行的制表符分隔文件

id1 name1   ['a', 'b']  3.0 2.0 0.0 1.0

即一个 id、一个名称、一个包含一些字符串的列表和一系列 4 个 float 属性。我正在阅读这个文件作为

rdd = sc.textFile('myfile.tsv') \ 
    .map(lambda row: row.split('\t')) 
df = sqlc.createDataFrame(rdd, schema)

我将模式指定为

schema = StructType([ 
    StructField('id', StringType(), True), 
    StructField('name', StringType(), True), 
    StructField('list', ArrayType(StringType()), True), 
    StructField('att1', FloatType(), True), 
    StructField('att2', FloatType(), True), 
    StructField('att3', FloatType(), True), 
    StructField('att4', FloatType(), True) 
])

问题是，从 DataFrame 上的 collect 判断，列表和属性都没有被正确读取。事实上，我得到的都是 None:

Row(id=u'id1', brand_name=u'name1', list=None, att1=None, att2=None, att3=None, att4=None)

我做错了什么？

请您参考如下方法:

它已被正确阅读，只是没有按您预期的那样工作。模式参数声明是什么类型以避免昂贵的模式推断而不是如何转换数据。提供与声明的架构匹配的输入是您的责任。

这也可以通过数据源处理(查看 spark-csv 和 inferSchema 选项)。但它不会处理像数组这样的复杂类型。

由于您的模式大部分是扁平的，并且您知道可以尝试这样的类型:

df = rdd.toDF([f.name for f in schema.fields]) 
 
exprs = [ 
    # You should excluding casting 
    # on other complex types as well 
    col(f.name).cast(f.dataType) if f.dataType.typeName() != "array"  
    else col(f.name) 
    for f in schema.fields 
] 
 
df.select(*exprs)

并使用字符串处理函数或 UDF 单独处理复杂类型。或者，由于您无论如何都是在 Python 中读取数据，因此只需在创建 DF 之前强制执行所需的类型即可。