python - PySpark:类型错误:StructType 无法接受类型 或 的对象

Question

我正在从 CSV 文件读取数据，然后创建一个 DataFrame。但是当我尝试访问 DataFrame 中的数据时，我收到 TypeError。

fields = [StructField(field_name, StringType(), True) for field_name in schema.split(',')]
schema = StructType(fields)

input_dataframe = sql_context.createDataFrame(input_data_1, schema)

print input_dataframe.filter(input_dataframe.diagnosis_code == '11').count()

“unicode”和“str”均不适用于 Spark DataFrame。我收到以下类型错误:

类型错误:StructType 无法接受类型中的对象类型错误:StructType 无法接受类型中的对象

我尝试按如下方式使用“utf-8”进行编码，但仍然收到错误，但现在提示“str”出现 TypeError:

input_data_2 = input_data_1.map(lambda x: x.encode("utf-8"))
input_dataframe = sql_context.createDataFrame(input_data_2, schema)

print input_dataframe.filter(input_dataframe.diagnosis_code == '410.11').count()

我还尝试使用参数 use_unicode=True/False 将 CSV 直接解析为 utf-8 或 unicode

Answer 1

阅读字里行间。你是

reading data from a CSV file

并得到

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

发生这种情况是因为您传递的字符串不是与结构兼容的对象。也许您传递的数据如下:

input_data_1 = sc.parallelize(["1,foo,2", "2,bar,3"])

和架构

schema = "x,y,z"

fields = [StructField(field_name, StringType(), True) for field_name in schema.split(',')]
schema = StructType(fields)

并且您希望 Spark 能够解决问题。但事实并非如此。你可以

input_dataframe = sqlContext.createDataFrame(input_data_1.map(lambda s: s.split(",")), schema)

但说实话，只需使用 Spark csv 阅读器:

spark.read.schema(schema).csv("/path/to/file")