pyspark 读取包含 json 和非 json 列的文件

Question

我正在尝试读取和转换包含 json 和非 json 列的 csv 文件。 我设法读取文件并将其放入数据框中。架构是这样的：

root
 |-- 'id': string (nullable = true)
 |-- 'score': string (nullable = true)

如果我这样做df.take(2)，我会得到以下结果：

[Row('id'=u"'AF03DCAB-EE3F-493A-ACD9-4B98F548E6F3'", 'score'=u"{'topSpeed':15.00000,'averageSpeed':5.00000,'harshBraking':0,'harshAcceleration':0,'driverRating':null,'idlingScore':70,'speedingScore':70,'brakingScore':70,'accelerationScore':70,'totalEcoScore':70 }"), Row('id'=u"'1938A2B9-5EF2-413C-A7A3-C5F324FD4089'", 'score'=u"{'topSpeed':106.00000,'averageSpeed':71.00000,'harshBraking':0,'harshAcceleration':0,'driverRating':9,'idlingScore':76,'speedingScore':87,'brakingScore':86,'accelerationScore':82,'totalEcoScore':83 }")]

id 列是“普通”列，score 列包含 json 格式的数据。 我想将 json 内容分成单独的列，但还需要 id 列和其余数据。 A 只有分数列的工作代码：

df = rawdata.select("'score'")
df1 = df.rdd  # Convert to rdd
df2 = df1.flatMap(lambda x: x)  # Flatten rows
dfJsonScore = sqlContext.read.json(df2)
dfJsonScore.printSchema()
dfJsonScore.take(3)

这给了我这个：

root
 |-- accelerationScore: long (nullable = true)
 |-- averageSpeed: double (nullable = true)
 |-- brakingScore: long (nullable = true)
 |-- driverRating: long (nullable = true)
 |-- harshAcceleration: long (nullable = true)
 |-- harshBraking: long (nullable = true)
 |-- idlingScore: long (nullable = true)
 |-- speedingScore: long (nullable = true)
 |-- topSpeed: double (nullable = true)
 |-- totalEcoScore: long (nullable = true)

[Row(accelerationScore=70, averageSpeed=5.0, brakingScore=70, driverRating=None, harshAcceleration=0, harshBraking=0, idlingScore=70, speedingScore=70, topSpeed=15.0, totalEcoScore=70),
 Row(accelerationScore=82, averageSpeed=71.0, brakingScore=86, driverRating=9, harshAcceleration=0, harshBraking=0, idlingScore=76, speedingScore=87, topSpeed=106.0, totalEcoScore=83),
 Row(accelerationScore=81, averageSpeed=74.0, brakingScore=85, driverRating=9, harshAcceleration=0, harshBraking=0, idlingScore=75, speedingScore=87, topSpeed=102.0, totalEcoScore=82)]

但我无法让它与 id 列结合使用。

Answer 1

有一个全新的from_json 函数added in pyspark 2.1 可以处理您的情况。

具有以下架构的数据框：

>>> df.printSchema()
root
 |-- id: string (nullable = true)
 |-- score: string (nullable = true)

首先为 json 字段生成架构：

>>> score_schema = spark.read.json(df.rdd.map(lambda row: row.score)).schema

然后在from_json中使用它：

>>> df.withColumn('score', from_json('score', score_schema)).printSchema()
root
 |-- id: string (nullable = true)
 |-- score: struct (nullable = true)
 |    |-- accelerationScore: long (nullable = true)
 |    |-- averageSpeed: double (nullable = true)
 |    |-- brakingScore: long (nullable = true)
 |    |-- driverRating: long (nullable = true)
 |    |-- harshAcceleration: long (nullable = true)
 |    |-- harshBraking: long (nullable = true)
 |    |-- idlingScore: long (nullable = true)
 |    |-- speedingScore: long (nullable = true)
 |    |-- topSpeed: double (nullable = true)
 |    |-- totalEcoScore: long (nullable = true)

编辑

如果您不能使用 spark 2.1，get_json_object 始终是一个选项，但要求字段是有效的 json，即将" 作为字符串分隔符而不是 '，请参见此示例：

df.withColumn('score', regexp_replace('score', "'", "\"")) \
    .select(
        'id', 
        get_json_object('score', '$.accelerationScore').alias('accelerationScore'), 
        get_json_object('score', '$.topSpeed').alias('topSpeed')
    ).show()

+--------------------+-----------------+--------+
|                  id|accelerationScore|topSpeed|
+--------------------+-----------------+--------+
|AF03DCAB-EE3F-493...|               70|    15.0|
|1938A2B9-5EF2-413...|               82|   106.0|
+--------------------+-----------------+--------+