python - 我将如何在 Dask 中进行 Spark 爆炸？

Question

我是 dask 的新手，请多多包涵。

我有一个 JSON 文件，其中每一行都有以下架构:

{
 'id': 2,
 'version': 7.3,
 'participants': range(10)
}

participants 是一个嵌套字段。

input_file = 'data.json'   
df = db.read_text(input_file).map(json.loads)

我可以做:
df.pluck(['id', 'version'])
或者
df.pluck('参与者').flatten()

但是我怎样才能做相当于 Spark 爆炸的事情，我可以同时选择 id、version 并展平 participants ?

所以输出将是:

{'id': 2, 'version': 7.3, 'participants': 0}
{'id': 2, 'version': 7.3, 'participants': 1}
{'id': 2, 'version': 7.3, 'participants': 2}
{'id': 2, 'version': 7.3, 'participants': 3}
...

Answer 1

可以编写自定义函数，使用 dask.bag.from_sequence 读取和转换文件的行

def mapper(row, denest_field):
    js = json.loads(row)
    for v in js[denest_field]:
        yield {'id': js['id'], denest_field: v, 'version': js['version']}


def yield_unnested(fname, denest_field):
    with open (fname) as f:
        for row in f:
            yield from mapper(row, denest_field)

我保存了一个名为 'data.json' 的文件，其中包含以下内容

{"id": 2, "version": 7.3, "participants": [0,1,2,3,4,5,6,7,9,9]}

然后用from_sequence读取

df = db.from_sequence(yield_unnested('data.json', 'participants'))
list(df) # outputs:

[{'id': 2, 'participants': 0, 'version': 7.3},
 {'id': 2, 'participants': 1, 'version': 7.3},
 {'id': 2, 'participants': 2, 'version': 7.3},
 {'id': 2, 'participants': 3, 'version': 7.3},
 {'id': 2, 'participants': 4, 'version': 7.3},
 {'id': 2, 'participants': 5, 'version': 7.3},
 {'id': 2, 'participants': 6, 'version': 7.3},
 {'id': 2, 'participants': 7, 'version': 7.3},
 {'id': 2, 'participants': 9, 'version': 7.3},
 {'id': 2, 'participants': 9, 'version': 7.3}]

请注意，我是 dask 的新手，这可能不是处理问题的最有效方法。