带有缩进的 Pyspark textFile json答案

【问题标题】：Pyspark textFile json with indentation带有缩进的 Pyspark textFile json
【发布时间】：2016-02-04 07:54:42
【问题描述】：

Wanne 将带有缩进的 json 文件读入 RDD，但 spark 会抛出异常。

# txts = sc.textFile('data/jsons_without_indentation') # works
txts = sc.textFile('data/jsons_with_indentation')      # fails
txts_dicts = txts.map(lambda data: json.loads(data))
txts_dicts.collect()

sc.wholeTextFiles 也不起作用。是否可以在不先将其转换为文件的情况下加载带有缩进的 json？

示例 json 文件如下所示：

{
    "data": {
        "text": {
            "de": "Ein Text.",
            "en": "A text."
        }
    }
}

【问题讨论】：

所以文件是由多个json组成的，每个都在多行中？
不，它是一个json文件的文件夹，每个文件里面的json的缩进级别为4。Spark不喜欢它，不管是在一个文件中还是在多个文件中。每行一个 json - jsonl 开箱即用 - 毫无疑问。
你能添加一个文件/行的例子吗？
抛出了哪个异常？

标签： python json apache-spark pyspark

【解决方案1】：

如果这只是每个文件的单个 JSON 文档，您只需要 SparkContext.wholeTextFiles。首先让我们创建一些虚拟数据：

import tempfile
import json 

input_dir = tempfile.mkdtemp()

docs = [
    {'data': {'text': {'de': 'Ein Text.', 'en': 'A text.'}}},
    {'data': {'text': {'de': 'Ein Bahnhof.', 'en': 'A railway station.'}}},
    {'data': {'text': {'de': 'Ein Hund.', 'en': 'A dog.'}}}]

for doc in docs:
    with open(tempfile.mktemp(suffix="json", dir=input_dir), "w") as fw:
        json.dump(doc, fw, indent=4)

现在让我们读取数据：

rdd = sc.wholeTextFiles(input_dir).values()

并确保文件确实是缩进的：

print rdd.top(1)[0]

## {
##     "data": {
##         "text": {
##             "de": "Ein Text.", 
##             "en": "A text."
##         }
##     }
## }

终于可以解析了：

parsed = rdd.map(json.loads)

并检查一切是否按预期工作：

parsed.takeOrdered(3)

## [{u'data': {u'text': {u'de': u'Ein Bahnhof.', u'en': u'A railway station.'}}},
##  {u'data': {u'text': {u'de': u'Ein Hund.', u'en': u'A dog.'}}},
##  {u'data': {u'text': {u'de': u'Ein Text.', u'en': u'A text.'}}}]

如果您仍然遇到一些问题，很可能是由于某些格式错误的条目。您可以做的最简单的事情是使用带有自定义包装器的flatMap 丢弃格式错误的条目：

rdd_malformed = sc.parallelize(["{u'data': {u'text': {u'de':"]).union(rdd)

## org.apache.spark.api.python.PythonException: Traceback (most recent call ...
##     ...
## ValueError: Expecting property name: line 1 column 2 (char 1)

并使用try_seq 包装（此处定义：What is the equivalent to scala.util.Try in pyspark?）

rdd_malformed.flatMap(lambda x: seq_try(json.loads, x)).collect()

## [{u'data': {u'text': {u'de': u'Ein Hund.', u'en': u'A dog.'}}},
##  {u'data': {u'text': {u'de': u'Ein Text.', u'en': u'A text.'}}},
##  {u'data': {u'text': {u'de': u'Ein Bahnhof.', u'en': u'A railway station.'}}}]

【讨论】：