标签: dask.distributed

我正在尝试将Wikipedia CirrusSearch 转储转换为由 450G 16 核 GCP 实例上的标题索引的 Parquet 支持的 dask 数据帧。CirrusSearch 转储以单个 json 行格式文件的形式出现。英文 Wipedia 转储包含 500 万个 recard，压缩为 12G，扩展为 90+G。一个重要的细节是记录并不完全平坦。

最简单的方法是

import json
import dask
from  dask import bag as db, dataframe as ddf
from  toolz import curried as tz
from toolz.curried import operator as op

blocksize=2**24
npartitions='auto'
parquetopts=dict(engine='fastparquet', object_encoding='json')

lang = 'en'
wiki = 'wiki'
date = 20180625
path='./'

source = f'{path}{lang}{wiki}-{date}-cirrussearch-content.json'

(
 db
 .read_text(source, blocksize=blocksize)
 .map(json.loads)
 .filter(tz.flip(op.contains, 'title'))
 .to_dataframe()
 .set_index('title', npartitions=npartitions)
 .to_parquet(f'{lang}{wiki}-{date}-cirrussearch.pq', **parquetopts)
) …