我想写入一个 gs 文件，但我在编译时不知道文件名。它的名称基于在运行时定义的行为。我该如何继续？
【问题讨论】：
标签： apache-beam
如果您使用的是 Beam Java，您可以为此使用 FileIO.writeDynamic()（从目前正在发布的 Beam 2.3 开始 - 但您已经可以通过版本 2.3.0-SNAPSHOT 使用它），或者较旧的 DynamicDestinations API（在 Beam 2.2 中可用）。
使用FileIO.writeDynamic() 将银行交易PCollection 写入GCS 上不同路径的示例，具体取决于交易类型：
PCollection<BankTransaction> transactions = ...;
transactions.apply(
FileIO.<BankTransaction, TransactionType>writeDynamic()
.by(Transaction::getType)
.via(BankTransaction::toString, TextIO.sink())
.to("gs://bucket/myfolder/")
.withNaming(type -> defaultNaming("transactions_", ".txt"));
有关DynamicDestinations 使用的示例，请参阅example code in the TextIO unit tests。
或者，如果您想将每条记录写入其自己的文件，只需使用来自DoFn 的FileSystems API（特别是FileSystems.create()）。
【讨论】：
.via(___, ___) 行，那么它看起来需要将第一个参数包装在Contextful.fn(___) 中
对于 Python 人群：
2.14.0 的 Beam python SDK 中添加了一个实验性写入，beam.io.fileio.WriteToFiles：
my_pcollection | beam.io.fileio.WriteToFiles(
path='/my/file/path',
destination=lambda record: 'avro' if record['type'] == 'A' else 'csv',
sink=lambda dest: AvroSink() if dest == 'avro' else CsvSink(),
file_naming=beam.io.fileio.destination_prefix_naming())
它可用于每条记录写入不同的文件。
如果您的文件名基于 pcollections 中的数据，您可以使用 destination 和 file_naming 根据每条记录的数据创建文件。
更多文档在这里：
https://beam.apache.org/releases/pydoc/2.14.0/apache_beam.io.fileio.html#dynamic-destinations
还有这里的 JIRA 问题：
【讨论】：
正如@anrope 已经提到的，apache_beam.io.fileio 似乎是用于写入文件的最新 Python API。 WordCount 示例目前已过时，因为它使用 WriteToText 类，该类继承自现已弃用的 apache_beam.io.filebasedsink / apache_beam.io.iobase
要添加到现有答案，这是我的管道，我在运行时动态命名输出文件。我的管道接受 N 个输入文件并创建 N 个输出文件，这些文件根据其对应的输入文件名命名。
with beam.Pipeline(options=pipeline_options) as p:
(p
| 'CreateFiles' >> beam.Create(input_file_paths)
| 'MatchFiles' >> MatchAll()
| 'OpenFiles' >> ReadMatches()
| 'LoadData' >> beam.Map(custom_data_loader)
| 'Transform' >> beam.Map(custom_data_transform)
| 'Write' >> custom_writer
)
当我加载数据时，我创建了一个元组记录(file_name, data) 的 PCollection。我的所有转换都应用于data，但我将file_name 传递到管道的末尾以生成输出文件名。
def custom_data_loader(f: beam.io.fileio.ReadableFile):
file_name = f.metadata.path.split('/')[-1]
data = custom_read_function(f.open())
return file_name, data
def custom_data_transform(record):
file_name, data = record
data = custom_transform_function(data) # not defined
return file_name, data
我将文件保存为：
def file_naming(record):
file_name, data = record
file_name = custom_naming_function(file_name) # not defined
return file_name
def return_destination(*args):
"""Optional: Return only the last arg (destination) to avoid sharding name format"""
return args[-1]
custom_writer = WriteToFiles(
path='path/to/output',
file_naming=return_destination,
destination=file_naming,
sink=TextSink()
)
用您自己的逻辑替换所有 custom_* 函数。
【讨论】：
我知道这是一个老问题，但我对文档中的示例感到困惑。
这里是一个简单的例子，说明如何根据 dict 项拆分文件。
pipeline_options = PipelineOptions()
pipeline_options.view_as(SetupOptions).save_main_session = False
def file_names(*args):
file_name = fileio.destination_prefix_naming()(*args)
destination, *_ = file_name.split("----")
return f"{destination}.json"
class JsonSink(fileio.TextSink):
def write(self, element):
record = json.loads(element)
record.pop("id")
self._fh.write(json.dumps(record).encode("utf8"))
self._fh.write("\n".encode("utf8"))
def destination(element):
return json.loads(element)["id"]
with beam.Pipeline(options=pipeline_options) as p:
data = [
{"id": 0, "message": "whhhhhhyyyyyyy"},
{"id": 1, "message": "world"},
{"id": 1, "message": "hi there!"},
{"id": 1, "message": "what's up!!!?!?!!?"},
]
(
p
| "CreateEmails" >> beam.Create(data)
| "JSONify" >> beam.Map(json.dumps)
| "Write Files"
>> fileio.WriteToFiles(
path="path/",
destination=destination,
sink=lambda dest: JsonSink(),
file_naming=file_names,
)
)
【讨论】：