如何在Python中将JSON文件目录加载到Apache Spark中

Question

我对Apache Spark还比较新，我想在Python中从包含多个JSON文件的字典列表中创建一个单一的RDD（每个文件都经过了gzip压缩并包含一个字典列表）。大致上来说，生成的RDD将包含所有字典列表组合成的单个字典列表。我没有在文档（https://spark.apache.org/docs/1.2.0/api/python/pyspark.html）中找到这个问题的解决方法，如果我错过了，请告诉我。

到目前为止，我尝试读取JSON文件并在Python中创建组合列表，然后使用sc.parallelize()，但是整个数据集太大而无法全部加载到内存中，因此这不是实际可行的解决方案。看起来Spark应该有一种聪明的方式来处理这种情况，但我不知道它是什么。

如何在Python中创建一个包括所有JSON文件中列表的单一RDD？

我还要提到，我不想使用Spark SQL。如果可能的话，我想使用像map、filter等函数。

- Brandt

Answer 1

根据tgpfeiffer在他们的回答和评论中提到的，这是我所做的。

首先，正如他们所提到的，必须对JSON文件进行格式化，使它们每行只有一个字典，而不是一个字典列表。然后，问题就很简单了：

my_RDD_strings = sc.textFile(path_to_dir_with_JSON_files)
my_RDD_dictionaries = my_RDD_strings.map(json.loads)

如果有更好或更有效的方法，请告诉我，但这种方法似乎有效。

- Brandt

Answer 2

您可以使用 sqlContext.jsonFile() 方法获取一个 SchemaRDD（即 RDD[Row] 加上一个 schema），用于与 Spark SQL 一起使用。或者，参见 Loading JSON dataset into Spark, then use filter, map, etc 获取非 SQL 处理流水线。我想您可能需要解压文件，并且 Spark 只能处理每行都是单个 JSON 文档的文件（即不支持多行对象）。

- tgpfeiffer

---

感谢您的回答。我应该提到我不想使用Spark SQL，我想使用像您所引用的问题中那样的非SQL处理流程。我将更新我的原始问题。您所提到的问题的答案似乎是用Scala编写的，而不是Python。尽管如此，再次感谢您的帮助！ - Brandt

没错，这是用Scala编写的，但是这个思路可以应用到你的问题上：使用sparkContext.textFile()加载输入数据集（实际上似乎支持gzipped文件），然后使用你选择的解析器（例如json模块）解析字符串行，然后按照你的意愿进行处理。 - tgpfeiffer

谢谢，这个可行！关键步骤是在 json.loads 上使用 map 函数。我将会把我所做的完整过程作为答案发布。非常感谢您的帮助。 - Brandt

Answer 3

从文件中加载Json列表作为RDD：

```html

使用RDD：

```

def flat_map_json(x): return [each for each in json.loads(x[1])]   
rdd = sc.wholeTextFiles('example.json').flatMap(flat_map_json)

- Supritha P

Answer 4

你可以使用textFile将一个文件夹中的文件加载到单个RDD中，它还支持通配符。这不会给你文件名，但你似乎也不需要它们。

在使用基本转换（如map、filter等）时，你可以使用Spark SQL。SchemaRDD也是一个RDD（在Python和Scala中都是如此）。

- pzecevic