Pyspark textFile json with indentation

Wanne将一个带有缩进的json文件读入RDD，但spark会引发异常。Pyspark textFile json with indentation

# 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." 
     } 
    } 
}

来源

2015-11-03 rebeling

因此该文件是由多个'json'的每个人都有多行？ – Udy

不，它是json文件的文件夹，每个文件的json内部都有一个4的缩进级别.Spark不喜欢它，既不在一个也不在多个文件中。每行一个json - jsonl开箱即用 - 毫无疑问。 – rebeling

你能添加一个文件/行的例子吗？ – Udy

如果这仅仅是每个文件一个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)

and check如果一切都按预期：

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.'}}}]

来源

2015-11-09 23:42:42 zero323

Pyspark textFile json with indentation

回答

相关问题