处理Spark数据帧中的非均匀JSON列 [英] Dealing with non-uniform JSON columns in spark dataframe

问题描述

我想知道将换行符分隔的JSON文件读入数据帧的最佳实践是什么.至关重要的是，每个记录中的(必填)字段之一映射到一个不能保证具有相同子字段的对象(即，架构在所有记录中都是不一致的).

I would like to know what is the best practice for reading a newline delimited JSON file into a dataframe. Critically, one of the (required) fields in each record maps to an object that is not guaranteed to have the same sub-fields (ie the schema is non-uniform across all the records).

例如，输入文件可能看起来像:

For example, an input file might look like:

{"id": 1, "type": "foo", "data": {"key0": "foo", "key2": "meh"}}
{"id": 2, "type": "bar", "data": {"key2": "poo", "key3": "pants"}}
{"id": 3, "type": "baz", "data": {"key3": "moo"}}

在这种情况下，字段id，type和data将出现在所有记录中，但是data映射到的结构将具有异构模式.

In this case the fields id, type, and data will be present in all records, but the struct mapped to by data will have a heterogeneous schema.

我有两种方法来解决data列的不一致性:

I have two approaches for dealing wi the non-uniformity of the data column:

1. 让spark推断模式:

df = spark.read.options(samplingRatio=1.0).json('s3://bucket/path/to/newline_separated_json.txt')

此方法的一个明显问题是需要对每条记录进行采样以确定最终的模式的字段/方案的超集.给定一个数据集的记录数以亿计的低值，这可能会过高地花费?或者...

The obvious issue with this approach is the need to sample every record to determine the super-set of fields/schemas that will be the final schema. This may be prohibitively expensive given a dataset in the low 100s of millions of records? Or...

1. 告诉我们将数据字段强制转换为JSON字符串，然后仅拥有一个由三个顶级字符串字段(id，type，data)组成的架构.在这里，我不太确定最好的进行方法.例如，我假设仅声明data字段为字符串，如下所示，将不起作用，因为它没有明确地执行json.dumps的等效功能?

1. Tell spark to coerce the data field into a JSON string, and then just have a schema comprised of three top-level string fields, id, type, data. And here I'm not really sure the best way to proceed. For example I am assuming just declaring the data field to be a string as in the following, will not work because it is not explicitly doing the equivalent of json.dumps?

schema = StructType([
    StructField("id", StringType(), true),
    StructField("type", StringType(), true),
    StructField("data", StringType(), true)
])
df = spark.read.json('s3://bucket/path/to/newline_separated_json.txt', schema=schema)

如果我想避免扫描由选项1引起的完整数据集的开销，什么是摄取此文件并将data字段保留为JSON字符串的最佳方法?

If I want to avoid the cost of scanning the full dataset incurred by option 1, what is the best way to ingest this file and keep the data field as a JSON string?

谢谢

推荐答案

我认为您的尝试和总体构想都朝着正确的方向发展.这是另外两种基于数据帧API的内置选项aka get_json_object/from_json的方法，并通过RDD API将map转换与python的json.dumps()和json.loads()一起使用.

I think your attempt and the overall idea is in the right direction. Here are two more approaches based on the build-in options aka get_json_object/from_json via dataframe API and using map transformation along with python's json.dumps() and json.loads() via the RDD API.

选项1: get_json_object()/ from_json()

首先，我们尝试使用不需要架构的get_json_object():

First let's try with get_json_object() which doesn't require a schema:

import pyspark.sql.functions as f

df = spark.createDataFrame([
  ('{"id": 1, "type": "foo", "data": {"key0": "foo", "key2": "meh"}}'),
  ('{"id": 2, "type": "bar", "data": {"key2": "poo", "key3": "pants"}}'),
  ('{"id": 3, "type": "baz", "data": {"key3": "moo"}}')
], StringType())

df.select(f.get_json_object("value", "$.id").alias("id"), \
          f.get_json_object("value", "$.type").alias("type"), \
           f.get_json_object("value", "$.data").alias("data"))

# +---+----+-----------------------------+
# |id |type|data                         |
# +---+----+-----------------------------+
# |1  |foo |{"key0":"foo","key2":"meh"}  |
# |2  |bar |{"key2":"poo","key3":"pants"}|
# |3  |baz |{"key3":"moo"}               |
# +---+----+-----------------------------+

相反，from_json()需要架构定义:

from pyspark.sql.types import StringType, StructType, StructField
import pyspark.sql.functions as f

df = spark.createDataFrame([
  ('{"id": 1, "type": "foo", "data": {"key0": "foo", "key2": "meh"}}'),
  ('{"id": 2, "type": "bar", "data": {"key2": "poo", "key3": "pants"}}'),
  ('{"id": 3, "type": "baz", "data": {"key3": "moo"}}')
], StringType())

schema = StructType([
    StructField("id", StringType(), True),
    StructField("type", StringType(), True),
    StructField("data", StringType(), True)
])

df.select(f.from_json("value", schema).getItem("id").alias("id"), \
         f.from_json("value", schema).getItem("type").alias("type"), \
         f.from_json("value", schema).getItem("data").alias("data"))

# +---+----+-----------------------------+
# |id |type|data                         |
# +---+----+-----------------------------+
# |1  |foo |{"key0":"foo","key2":"meh"}  |
# |2  |bar |{"key2":"poo","key3":"pants"}|
# |3  |baz |{"key3":"moo"}               |
# +---+----+-----------------------------+

选项2:map/RDD API + json.dumps()

from pyspark.sql.types import StringType, StructType, StructField
import json

df = spark.createDataFrame([
  '{"id": 1, "type": "foo", "data": {"key0": "foo", "key2": "meh"}}',
  '{"id": 2, "type": "bar", "data": {"key2": "poo", "key3": "pants"}}',
  '{"id": 3, "type": "baz", "data": {"key3": "moo"}}'
], StringType())

def from_json(data):
  row = json.loads(data[0])
  return (row['id'], row['type'], json.dumps(row['data']))

json_rdd = df.rdd.map(from_json)

schema = StructType([
    StructField("id", StringType(), True),
    StructField("type", StringType(), True),
    StructField("data", StringType(), True)
])

spark.createDataFrame(json_rdd, schema).show(10, False)

# +---+----+--------------------------------+
# |id |type|data                            |
# +---+----+--------------------------------+
# |1  |foo |{"key2": "meh", "key0": "foo"}  |
# |2  |bar |{"key2": "poo", "key3": "pants"}|
# |3  |baz |{"key3": "moo"}                 |
# +---+----+--------------------------------+

函数from_json会将字符串行转换为(id, type, data)的元组. json.loads()将解析json字符串并返回一个字典，通过该字典我们可以生成并返回最后的元组.

Function from_json will transform the string row into a tuple of (id, type, data). json.loads() will parse the json string and return a dictionary through which we generate and return the final tuple.

这篇关于处理Spark数据帧中的非均匀JSON列的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！