标签: apache-beam

代码如下所示:

return pcol.apply(ParDo.named("FindTheBug")
        .withSideInputs(foo)
        .withSideInputs(bar(
        .of(new DoFn<T, U>() {
            F myFoo = c.sideInput(foo);
            B myBar = c.sideInput(bar);

声明这些侧输入,为什么Dataflow没有看到它们？

每个用于数据流的Google文档都说它现在基于Apache Beam，将我定向到Beam网站真是令人困惑。另外，如果我寻找github项目，我会看到google dataflow项目为空，而所有内容都将归入Apache Beam Repo。现在说我需要创建一个管道，根据从Apache Beam读取的内容，我会做的：from apache_beam.options.pipeline_options但是，如果我使用google-cloud-dataflow，则会出现错误：no module named 'options'，事实证明我应该使用from apache_beam.utils.pipeline_options。因此，看起来google-cloud-dataflow具有较旧的Beam版本，将被弃用吗？

我应该选择哪一个来开发我的数据流管道？

我有一个Apache Beam管道成功运行作为数据流模板.但是,我有一个选项类(扩展DataflowPipelineOptions).生成模板时,此类从pom或命令行界面获取参数.我想知道是否存在某个类,以便我可以直接从属性文件中加载这些参数.这样,从环境切换到环境会更容易,而且更干净

我想写一个gs文件,但我不知道编译时的文件名.它的名称基于在运行时定义的行为.我该怎么办？

因此，我正在做一个小项目，该项目使用Google Dataflow和apache beam建立流传输管道。我完成了一些教程，能够建立流水线并在BigQuery中运行流，但是我想将其流到完整的关系数据库（即Cloud SQL）中。我已经在这个网站和整个google上进行了搜索，似乎最好的方法是使用JdbcIO。我在这里有点困惑，因为当我查找有关如何执行此操作的信息时，它们都是指批量写入云SQL而不是完全流式传输。

我的简单问题是，我可以将数据直接流到Cloud SQL中，还是必须通过批处理发送。

干杯!

在阅读有关使用Java处理apache Beam中的流式元素的知识时，我遇到了DoFn<InputT, OutputT>，然后发现SimpleFunction<InputT, OutputT>。

两者看上去都与我相似，我很难理解它们之间的区别。

有人可以解释外行术语的区别吗？

给定如下数据集

{"slot":"reward","result":1,"rank":1,"isLandscape":false,"p_type":"main","level":1276,"type":"ba","seqNum":42544}
{"slot":"reward_dlg","result":1,"rank":1,"isLandscape":false,"p_type":"main","level":1276,"type":"ba","seqNum":42545}

我尝试通过过滤这些json数据type:ba并将其插入到使用python sdk的bigquery中

ba_schema = 'slot:STRING,result:INTEGER,play_type:STRING,level:INTEGER'

class ParseJsonDoFn(beam.DoFn):
    B_TYPE = 'tag_B'
    def process(self, element):
        text_line = element.trip()
        data = json.loads(text_line)

        if data['type'] == 'ba':
            ba = {'slot': data['slot'], 'result': data['result'], 'p_type': data['p_type'], 'level': data['level']}
            yield pvalue.TaggedOutput(self.B_TYPE, ba)

def run():
    parser = argparse.ArgumentParser()
    parser.add_argument('--input',
                      dest='input',
                      default='data/path/data',
                      help='Input file to process.')
    known_args, pipeline_args = parser.parse_known_args(argv)
    pipeline_args.extend([
      '--runner=DirectRunner',
      '--project=project-id',
      '--job_name=data-job',
    ])
    pipeline_options = PipelineOptions(pipeline_args)
    pipeline_options.view_as(SetupOptions).save_main_session = True
    with beam.Pipeline(options=pipeline_options) as p:
        lines = p | ReadFromText(known_args.input)

        multiple_lines = ( …

我正在创建一个Shell脚本来处理某些工作流程的自动化，该工作流程包括通过Apache Beam GCP访问Google Buckets。我正在使用我的服务帐户使用.json文件，在这种情况下我需要使用：

 gcloud auth activate-service-account --key-file myfile.json

与

export GOOGLE_APPLICATION_CREDENTIALS=myfile.json

我的目标是能够访问Google Beam在Apache Beam（数据流）中记录和设置的PubSub消息发布时间。

    PCollection<PubsubMessage> pubsubMsg
            = pipeline.apply("Read Messages From PubSub",
            PubsubIO.readMessagesWithAttributes()
                .fromSubscription(pocOptions.getInputSubscription()));

似乎不包含一个作为属性。我试过了

 .withTimestampAttribute("publish_time")

也没有运气。我想念什么？是否可以在数据流中提取Google PubSub发布时间？

我正在尝试构建将Pub / Sub流式传输到BigQuery的以下示例：

https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/master/src/main/java/com/google/cloud/teleport/templates/PubSubToBigQuery.java

代码是：

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF …