为 WritetoFiles 设置文件名答案

【问题标题】：Set filename for WritetoFiles为 WritetoFiles 设置文件名
【发布时间】：2019-12-23 15:20:55
【问题描述】：

我的流程将文件存储在磁盘上，我需要设置文件名以便找回内容。

默认命名是窗口时间戳和计数器，这对我没有帮助。文档对我来说不够清楚。 (https://beam.apache.org/releases/pydoc/2.16.0/apache_beam.io.fileio.html?highlight=default_file_naming)

fileio.WriteToFiles(archive_storage, file_naming=beam.io.fileio.destination_prefix_naming())

我想将文件命名为 <HASH>.json，其中 HASH 是文件内数据的名称。

【问题讨论】：

标签： python-3.x google-cloud-dataflow apache-beam

【解决方案1】：

感谢this example，我能够得到一个有效的sn-p。在这种情况下，我们将根据每个记录的哈希为每个记录指定不同的destination，因为我们希望将每个元素写入不同的文件。此外，我们将传递名为hash_naming 的自定义命名函数：

data = [{'id': 0, 'message': 'hello'},
        {'id': 1, 'message': 'world'}]

(p
  | 'Create Events' >> beam.Create(data) \
  | 'JSONify' >> beam.Map(json.dumps) \
  | 'Print Hashes' >> beam.ParDo(PrintHashFn()) \
  | 'Write Files' >> fileio.WriteToFiles(
      path='./output',
      destination=lambda record: hash(record),
      sink=lambda dest: JsonSink(),
      file_naming=hash_naming))

在PrintHashFn 中，我们将使用每个哈希记录每个元素：

logging.info("Element: %s with hash %s", element, hash(element))

因此，对于我们的数据，我们将获得：

INFO:root:Element: {"message": "hello", "id": 0} with hash -1885604661473532601
INFO:root:Element: {"message": "world", "id": 1} with hash 9144125507731048840

可能有更好的方法，但我发现调用 fileio.destination_prefix_naming()(*args) 我们可以从默认命名方案 (-1885604661473532601----00000-00001) 中检索目标 (-1885604661473532601)：

def hash_naming(*args):
  file_name = fileio.destination_prefix_naming()(*args)  # -1885604661473532601----00000-00001
  destination = file_name.split('----')[0]  # -1885604661473532601
  return '{}.json'.format(destination)  # -1885604661473532601.json

请注意，如果您在混合中添加窗口，则获取子字符串的拆分可能会有所不同。

使用 2.16.0 SDK 和 DirectRunner 运行脚本，我得到以下输出：

$ ls output/
-1885604661473532601.json  9144125507731048840.json
$ cat output/-1885604661473532601.json 
"{\"message\": \"hello\", \"id\": 0}"

更新完整代码here。

【讨论】：

我取出了JsonSink()，不知道该怎么办。这会对文件名进行哈希处理，这很好，但我无法用它找回我的数据，我的目标是对文件中的数据进行哈希处理。我会改进我的问题以明确这一点。
是的，如果我理解正确的话，这就是我想要展示的。无论如何，我用几个小的 JSON 元素作为数据对其进行了修改，并且只将记录的哈希值传递给文件名。例如，{"message": "hello", "id": 0} 具有哈希 -1885604661473532601，因此它将被写入 -1885604661473532601.json。关于JsonSink()我只是从另一个例子中拿来的，记住你可以在这里查看完整的代码：gist.github.com/gxercavins/…