python - 将包含多个表的 250GB JSON 文件拆分为 Parquet

Question

我有一个具有以下示例格式的 JSON 文件，

{
    "Table1": {
        "Records": [
            {
                "Key1Tab1": "SomeVal",
                "Key2Tab1": "AnotherVal"
            },
            {
                "Key1Tab1": "SomeVal2",
                "Key2Tab1": "AnotherVal2"
            }
        ]
    },
    "Table2": {
        "Records": [
            {
                "Key1Tab1": "SomeVal",
                "Key2Tab1": "AnotherVal"
            },
            {
                "Key1Tab1": "SomeVal2",
                "Key2Tab1": "AnotherVal2"
            }
        ]
    }
}

根键是来自 SQL 数据库的表名，其对应的值是行。我想将 JSON 文件拆分为单独的 Parquet 文件，每个文件代表一个表。IE。 Table1.parquet 和 Table2.parquet。

最大的问题是文件的大小阻止我将它加载到内存中。因此，我尝试使用 dask.bag 来适应文件的嵌套结构。

import dask.bag as db
from dask.distributed import Client
client = Client(n_workers=4)

lines = db.read_text("filename.json")

但是使用 lines.take(4) 评估输出显示 dask 无法正确读取新行。

('{\n', '    "Table1": {\n', '        "Records": [\n', '            {\n')

我曾尝试寻找特定问题的解决方案，但没有成功。

是否有可能使用 dask 解决拆分问题，或者是否有其他工具可以完成这项工作？

Answer 1

按照建议here试试 dask.dataframe.read_json()方法

这可能就足够了，但我不确定如果您没有足够的内存来将整个结果数据帧存储在内存中，它会如何表现。

import dask.dataframe as dd
from dask.distributed import Client

client = Client()

df = dd.read_json("filename.json")
df.to_parquet("filename.parquet", engine='pyarrow')

文档

• https://distributed.dask.org/en/latest/manage-computation.html#dask-collections-to-futures
• https://examples.dask.org/dataframes/01-data-access.html#Write-to-Parquet

---

如果 Dask 在单个系统上不以 block 的形式处理文件(它可能不会愉快地这样做，因为 JSON 以这种方式解析显然不友好......尽管不幸的是我无法访问我的测试系统为了验证这一点)并且系统内存无法处理这个巨大的文件，您可以通过创建一个大的交换文件来使用磁盘空间扩展系统内存。

请注意，这将创建一个约 300G 的文件(增加 count 字段以获得更多)并且与内存相比可能难以置信慢(但对于您的需要，特别是如果它是一次性的)。

# create and configure swapfile
dd if=/dev/zero of=swapfile.img bs=10M count=30000 status=progress
chmod 600 swapfile.img
mkswap swapfile.img
swapon swapfile.img
#
# run memory-greedy task
# ...
# ensure processes have exited
#
# disable and remove swapfile to reclaim disk space
swapoff swapfile.img  # may hang for a long time
rm swapfile.img