标签: apache-beam

根据以下文档，如果您没有明确指定触发器，您将获得如下描述的行为：

如果未指定，默认行为是在水印通过窗口末尾时首先触发，然后在每次有迟到数据时再次触发。

这种行为对于 FixedWindow 也是如此吗？例如，您会假设固定窗口应该具有在水印通过窗口结束后重复触发的默认触发器，并丢弃所有延迟数据，除非明确处理延迟数据。另外，在源代码中的何处可以看到触发器的定义，例如 FixedWindow 对象？

我正在学习 Apache Beam 并尝试实现类似于 distcp 的东西。我使用 FileIO.read().filepattern() 来获取输入文件，但是在使用 FileIO.write 写入时，文件有时会合并。

在作业执行之前知道分区计数是不可能的。

PCollection<MatchResult.Metadata> pCollection = pipeline.apply(this.name(), FileIO.match().filepattern(path()))
  .apply(FileIO.readMatches())
  .apply(name(), FileIO.<FileIO.ReadableFile>write()
        .via(FileSink.create())
        .to(path()));

接收器的代码

@AutoValue
public abstract static class FileSink implements FileIO.Sink<FileIO.ReadableFile> {

    private OutputStream outputStream;

    public static FileSink create() {
      return new AutoValue_FileIOOperator_FileSink();
    }

    @Override
    public void open(WritableByteChannel channel) throws IOException {
      outputStream = Channels.newOutputStream(channel);
    }

    @Override
    public void write(FileIO.ReadableFile element) throws IOException {
      try (final InputStream inputStream = Channels.newInputStream(element.open())) {
        IOUtils.copy(inputStream, outputStream);
      }
    }

    @Override
    public void flush() throws IOException …

从像 pub/sub 这样的无限源读取数据后，我正在应用窗口化。我需要将属于一个窗口的所有记录写入一个单独的文件。我在 Java 中找到了这个，但在 python 中找不到任何东西。

我有一个 .txt 文件，它具有 JSON 格式。我想读取、操作和重组文件（更改字段名称...） 如何使用 Apache Beam 在 Python 中执行此操作？

现在我只能使用 ParDo 获取类中的 RunTime 值，还有其他方法可以像在我的函数中一样使用运行时参数吗？

这是我现在得到的代码：

class UserOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument('--firestore_document',default='')

def run(argv=None):

    parser = argparse.ArgumentParser()

    pipeline_options = PipelineOptions()

    user_options = pipeline_options.view_as(UserOptions)

    pipeline_options.view_as(SetupOptions).save_main_session = True

    with beam.Pipeline(options=pipeline_options) as p:

        rows = (p 
        | 'Create inputs' >> beam.Create(['']) 
        | 'Call Firestore' >> beam.ParDo(
                CallFirestore(user_options.firestore_document)) 
        | 'Read DB2' >> beam.Map(ReadDB2))

我希望 user_options.firestore_document 无需执行 ParDo 即可在其他功能中使用

我有一个用例，我读入存储在谷歌云存储中的换行 json 元素并开始处理每个 json。在处理每个 json 时，我必须调用外部 API 来执行重复数据删除，无论该 json 元素之前是否被发现。我在做一个ParDo与DoFn每个JSON。

我还没有看到任何在线教程说明如何从 apache beam DoFnDataflow调用外部 API 端点。

我正在使用JAVABeam 的 SDK。我学习的一些教程解释了使用startBundle和FinishBundle但我不清楚如何使用它

我很难理解 GCP Dataflow/Apache Beam 和 Spring Cloud Dataflow 之间的差异。我想要做的是转向更原生的流数据处理解决方案，因此我们的开发人员可以更专注于开发核心逻辑而不是管理基础设施。

我们有一个现有的流解决方案，它由 Spring Cloud 数据流“模块”组成，我们可以独立迭代和部署，就像微服务一样，效果很好，但我们希望迁移到我们业务提供的 GCP 中的现有平台，需要我们使用 GCP Dataflow。在高层次上，解决方案很简单：

流 1：

Kafka Source (S0) -> Module A1 (Ingest) -> Module B1 (Map) -> Module C1 (Enrich) -> Module D1 (Split) -> Module E1 (Send output to Sink S1)

流 2：

Kafka Source (S1) -> Module A2 (Ingest) -> Module B2 (Persist to DB) -> Module B3 (Send Notifications through various channels)

根据我的理解，我们想要采用的解决方案应该是相同的，但是模块将成为 GCP Dataflow 模块，源/接收器将成为 GCP Pub/Sub 而不是 kafka。

我遇到的大多数文档都没有将 SCDF 和 …

我还是 Beam 的新手，但是您究竟如何从 GCS 存储桶中的 CSV 文件中读取数据？我基本上使用 Beam 将这些文件转换为 Pandas 数据帧，然后应用 sklearn 模型来“训练”这些数据。我见过的大多数示例都预先定义了标题，我希望这个 Beam 管道可以推广到标题肯定不同的任何文件。有一个名为beam_utils的库可以完成我想做的事情，但后来我遇到了这个错误：AttributeError: module 'apache_beam.io.fileio' has no attribute 'CompressionTypes'

代码示例：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# The error occurs in this import
from beam_utils.sources import CsvFileSource

options = {
    'project': 'my-project',
    'runner:': 'DirectRunner',
    'streaming': False
}

pipeline_options = PipelineOptions(flags=[], **options)

class Printer(beam.DoFn):
    def process(self, element):
        print(element)

with beam.Pipeline(options=pipeline_options) as p:  # Create the Pipeline with the specified options.

    data = (p
            | …

嗨，我有几个查询，我想使用 Apache Beam 依次运行和保存结果，我见过一些类似的问题，但找不到答案。我习惯于使用 Airflow 设计管道，而我对 Apache Beam 还是比较陌生。我正在使用 Dataflow 运行程序。这是我现在的代码：我希望 query2 仅在 query1 结果保存到相应表后运行。我如何链接它们？

    PCollection<TableRow> resultsStep1 = getData("Run Query 1",
            "Select * FROM basetable");

    resultsStep1.apply("Save Query1 data",
            BigQueryIO.writeTableRows()
                    .withSchema(BigQueryUtils.toTableSchema(resultsStep1.getSchema()))
                    .to("resultsStep1")
                    .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
    );

    PCollection<TableRow> resultsStep2 = getData("Run Query 2",
            "Select * FROM resultsStep1");

    resultsStep2.apply("Save Query2 data",
            BigQueryIO.writeTableRows()
                    .withSchema(BigQueryUtils.toTableSchema(resultsStep2.getSchema()))
                    .to("resultsStep2")
                    .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
    );

这是我的 getData 函数定义：

private PCollection<TableRow> getData(final String taskName, final String query) {
    return pipeline.apply(taskName,
            BigQueryIO.readTableRowsWithSchema()
                    .fromQuery(query)
                    .usingStandardSql()
                    .withCoder(TableRowJsonCoder.of()));
}

编辑（更新）：结果： You can’t sequence the completion of a …

需要一些帮助。我有一些从 Pub/Sub 读取并写入 GCS 中的批处理文件的琐碎任务，但是在使用 fileio.WriteToFiles 时遇到了一些困难

with beam.Pipeline(options=pipeline_options) as p:
  input = (p | 'ReadData' >> beam.io.ReadFromPubSub(topic=known_args.input_topic).with_output_types(bytes)
             | "Decode" >> beam.Map(lambda x: x.decode('utf-8'))
             | 'Parse' >> beam.Map(parse_json)
             | ' data w' >> beam.WindowInto(
                 FixedWindows(60),
                 accumulation_mode=AccumulationMode.DISCARDING
             ))

  event_data = (input
             | 'filter events' >> beam.Filter(lambda x: x['t'] == 'event')
             | 'encode et' >> beam.Map(lambda x: json.dumps(x))
             | 'write events to file' >> fileio.WriteToFiles(
                    path='gs://extention/ga_analytics/events/', shards=0))

窗口触发后我需要一个文件，但文件数等于来自 Pubsub 的消息数，有人可以帮助我吗？ 当前输出文件， 但我只需要一个文件。