使用 Google Cloud Dataflow 创建大型 CSV 数据答案

【问题标题】：Create Large CSV data using Google Cloud Dataflow使用 Google Cloud Dataflow 创建大型 CSV 数据
【发布时间】：2017-03-29 04:19:10
【问题描述】：

我需要创建一个带有标题的 ~ 20 亿条记录 的大型 csv 文件。使用独立脚本创建需要很长时间，但是由于记录不相关，我了解云数据流可以使其分布式旋转我选择的多个工作 GCE 机器。云数据流是否总是需要输入。在这里，我正在尝试以编程方式生成以下格式的数据

ItemId,   ItemQuantity, ItemPrice, Salevalue, SaleDate
item0001, 25          , 100      , 2500      , 2017-03-18
item0002, 50          , 200      , 10000     , 2017-03-25

注意

ItemId 可以后缀为 0001 到 9999 之间的任意随机数 ItemQuantity 可以是 (1 到 1000) 之间的随机值 ItemPrice 可以是 (1 到 100 之间的随机值) SaleValue = ItemQuantity*ItemPrice 日期在 2015-01-01 到2017-12-31

任何语言都可以。

继续问题Generating large file using Google Cloud Dataflow

【问题讨论】：

标签： csv google-cloud-dataflow

【解决方案1】：

目前，还没有一种非常优雅的方式来做到这一点。在 Python 中，你会这样做（Java 也是一样，只是语法发生了变化）：

def generate_keys():
  for i in range(2000):
    # Generate 2000 key-value pairs to shuffle
    yield (i, 0)

def generate_random_elements():
  for i in range(1000000):
    yield random_element()

p = beam.Pipeline(my_options)
(p 
 | beam.Create(['any']) 
 | beam.FlatMap(generate_keys)
 | beam.GroupByKey()
 | beam.FlatMap(generate_random_elements)
 | beam.WriteToText('gs://bucket-name/file-prefix'))

在generate_keys() 中，我们生成了 2000 个不同的密钥，然后我们运行 GroupByKey 以便将它们随机分配给不同的工作人员。我们需要这样做，因为 DoFn目前不能在多个工作人员之间拆分。（一旦实现SplittableDoFn，这将容易得多。

请注意，当 Dataflow 将结果写入接收器时，它通常会将它们分成不同的文件（例如 gs://bucket-name/file-prefix-0000-00001 等），因此您需要将这些文件压缩在一起。

此外，您可以使用 --num_workers 10，或在 Dataflow 中生成任意多个，或使用自动缩放。

【讨论】：