如果这仅仅是每个文件一个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.'}}}]
因此该文件是由多个'json'的每个人都有多行? – Udy
不,它是json文件的文件夹,每个文件的json内部都有一个4的缩进级别.Spark不喜欢它,既不在一个也不在多个文件中。每行一个json - jsonl开箱即用 - 毫无疑问。 – rebeling
你能添加一个文件/行的例子吗? – Udy