python - 将数据增量写入 parquet 文件

Question

要从 pandas 数据框写入 parquet，我正在执行以下操作:

df = pd.DataFrame(DATA)
table = pa.Table.from_pandas(df)
pq.write_table(table, 'DATA.parquet')

但是，如果我假设有 1B 行，那么它就不能很好地工作，而且它无法放入内存。在那种情况下，我将如何增量写入数据。例如，像这样的东西:

DATA = []
BACTCH_SIZE = 10000
with open('largefile.csv') as f:
    for num, line in enumerate(f):
        if (len(DATA) == BATCH_SIZE):
            pq.write_table(pa.Table.from_pandas(pd.DataFrame(DATA)), 'DATA.parquet')
            DATA = []
        DATA.append(line.split(','))

if DATA: pq.write_table(pa.Table.from_pandas(pd.DataFrame(DATA)), 'DATA.parquet')

但是，我相信上面的内容只会继续覆盖 parquet 文件。我怎样才能做相当于追加的事情？

Answer 1

Hadoop 不适用于追加。只需将每个批处理的新文件写入单个目录，几乎所有 Hadoop API 都应该能够读取所有 parquet 文件

BACTCH_SIZE = 10000
c = 0
with open('largefile.csv') as f:
    for num, line in enumerate(f):
        if len(DATA) == BATCH_SIZE:
            pq.write_table(pa.Table.from_pandas(pd.DataFrame(DATA)), 'DATA.{}.parquet'.format(c))
            DATA = []
            c += 1
        DATA.append(line.split(','))

这也是 Spark 写入数据的方式；每个执行者一个文件

但是如果你有一个很大的 csv，只需将它放在 HDFS 中，然后在它上面创建一个 Hive 表，然后从那里将它转换为 parquet。根本不需要 Pandas