使用 PySpark 将嵌套的 JSON 解析为 Spark DataFrame

Question

我真的很想在使用 PySpark-SQL 解析嵌套的 JSON 数据方面得到一些帮助。数据具有以下架构（空格是出于保密目的而进行的编辑...）

架构

root
 |-- location_info: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- restaurant_type: string (nullable = true)
 |    |    |
 |    |    |
 |    |    |-- other_data: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- other_data_1 string (nullable = true)
 |    |    |    |    |-- other_data_2: string (nullable = true)
 |    |    |    |    |-- other_data_3: string (nullable = true)
 |    |    |    |    |-- other_data_4: string (nullable = true)
 |    |    |    |    |-- other_data_5: string (nullable = true)
 |    |    |
 |    |    |-- latitude: string (nullable = true)
 |    |    |
 |    |    |
 |    |    |
 |    |    |
 |    |    |
 |    |    |-- longitude: string (nullable = true)
 |    |    |
 |    |    |
 |    |    |
 |    |    |-- timezone: string (nullable = true)
 |-- restaurant_id: string (nullable = true)

我的目标 我基本上想将数据放入以下数据框

restaurant_id | latitude | longtitude | timezone 

我试过了

• Databricks notebook tutorial

• 如下代码

dfj = spark.read.option("multiLine", False).json("/file/path")

result = dfj.select(col('restaurant_id'),
  explode(col('location_info')).alias('location_info') )

# SQL operation
result.createOrReplaceTempView('result')

subset_data = spark.sql(
'''
SELECT restaurant_id, location_info.latitude,location_info.longitude,location_info.timestamp  
FROM result

'''
).show()  

# Also tried this to read in
source_df_1 = spark.read.json(sc.wholeTextFiles("/file/path")
          .values()
          .flatMap(lambda x: x
                   .replace("{", "#!#")
                   .split("#!#")))

但奇怪的是，它只为第一个对象或餐厅 ID 提供了以下内容

+-------+-----------+------------+--------------------+
|restaurant_id|latitude|longitude|timestamp|
+-------+-----------+------------+--------------------+
| 25|2.0|-8.0|2020-03-06T03:00:...|
| 25|2.0|-8.0|2020-03-06T03:00:...|
| 25|2.0|-8.0|2020-03-06T03:00:...|
| 25|2.0|-8.0|2020-03-06T03:01:...|
| 25|2.0|-8.0|2020-03-06T03:01:...|
+-------+-----------+------------+--------------------+

我的研究表明，这可能与 JSON 文件的源结构方式有关。例如：

{}{
}{
}

因此不是多线或其他东西。想知道该怎么做？

非常感谢您的阅读，任何帮助将不胜感激。我知道我总是可以指望 SO 提供帮助

Answer 1

我可以通过阅读上面描述的 JSON 文件来解决这个问题，希望对您有所帮助！ ：

# Reading multiple files in the dir
source_df_1 = spark.read.json(sc.wholeTextFiles("file_path/*")                              
          .values()
          .flatMap(lambda x: x
                   .replace('{"restaurant_id','\n{"restaurant_id' ).split('\n')))


# explode here to have restaurant_id, and nested data
exploded_source_df_1 =  source_df_1.select(col('restaurant_id'),
  explode(col('location_info')).alias('location_info') )


# Via SQL operation : this will solve the problem for parsing 
exploded_source_df_1.createOrReplaceTempView('result_1')

subset_data_1 = spark.sql(
'''
SELECT restaurant_id, location_infos.latitude,location_infos.longitude,location_infos.timestamp 
from result_1
'''
).persist()

Answer 2

spark.read.json() 阅读器假定每个文本行有一个 json 对象。我不确定我是否遵循\n 的插入然后拆分...听起来可能文件格式错误？

也许有一个记录分隔符，例如您看不到的 \r。 linux 命令od -c <file name> |head -10 将帮助显示记录之间的字符。

如果架构是众所周知的，则提供该架构对象，这将减少进行架构推断的第一遍。例如。 schema.read.schema(schema).json('path to directory') 绝对让您的读取操作更快。将对象保存为 parquet 或 delta Lake 格式以获得更好的性能，您需要稍后对其进行查询。

Databricks 的 COPY INTO 或 cloudFiles 格式将加快提取速度/减少延迟。 https://docs.databricks.com/spark/latest/structured-streaming/auto-loader.html