使用 Pyspark 使用 Spark 读取巨大 Json 文件的第一行

Posted 2023-04-15

【中文标题】使用 Pyspark 使用 Spark 读取巨大 Json 文件的第一行

【英文标题】：Read first line of huge Json file with Spark using Pyspark

【发布时间】：2018-01-10 13:27:06

【问题描述】：

我对 Spark 还很陌生，为了自学，我一直在使用小型 json 文件，它们运行良好。我在 Spark 2.2.1 中使用 Pyspark 但是我不知道如何读取 单个数据行而不是整个 json 文件。我一直在寻找这方面的文档，但似乎非常稀缺。我必须处理一个大的（比我的 RAM 大）json 文件（***转储：https://archive.org/details/wikidata-json-20150316），并且希望以块或逐行的方式执行此操作。我认为 Spark 旨在做到这一点，但不知道如何做到这一点，当我以一种天真的方式请求前 5 个观察结果时，我的内存不足。我试过 RDD。

SparkRDD= spark.read.json("largejson.json").rdd
SparkRDD.take(5)

和数据框

SparkDF= spark.read.json("largejson.json")
SparkDF.show(5,truncate = False)

简而言之：

1) 我如何读取大型 JSON 文件的一小部分？ （显示前 5 个条目）

2) 我如何逐行过滤大型 JSON 文件以仅保留所需的结果？

另外：我不想为此工作预定义数据方案。

我一定是忽略了什么。

谢谢

编辑：在一些帮助下，我看到了第一个观察结果，但它本身已经太大了，无法在此处发布，所以我只在这里放一小部分。

[

    "id": "Q1",
    "type": "item",
    "aliases": 
        "pl": [
            "language": "pl",
            "value": "kosmos"
        , 
            "language": "pl",
            "value": "\\u015bwiat"
        , 
            "language": "pl",
            "value": "natura"
        , 
            "language": "pl",
            "value": "uniwersum"
        ],
        "en": [
            "language": "en",
            "value": "cosmos"
        , 
            "language": "en",
            "value": "The Universe"
        , 
            "language": "en",
            "value": "Space"
        ],
...etc

【问题讨论】：

文档的结构是什么？

请在问题中提供样本。我怀疑有人会为了回答您的问题而下载 4GB 的数据。了解结构对于了解故障的根源很重要。

这正是我自己的问题：我无法取样，因为我无法访问前几行，因为我的内存不足，因为我不知道如何只取几行行而不是一切。 :D。我很乐意给你一个样本......从而回答我的问题的最大部分：D

head largejson.json 或 with open("large.json") as fr: s = fr.read(1000) ？ :) 我怀疑这是一个单一的 JSON 文档，答案是不行的，你必须编写自己的解析/准备逻辑。

感谢 sn-p，我用它来掌握第一个观察结果的样子 :)（虽然单个条目本身已经很大）。所以...Spark 不允许您通过观察来处理大型 json 文件观察？这让我感到惊讶，它是如何处理大文件的呢？你不能把所有东西都放在记忆中，那是不现实的。

【参考方案1】：

这与Select only first line from files under a directory in pyspark非常相似

因此这样的事情应该可以工作：

def read_firstline(filename):
    with open(filename, 'rb') as f:
        return f.readline()

# files is a list of filenames
rdd_of_firstlines = sc.parallelize(files).flatMap(read_firstline)