标签: apache-beam

我正在尝试DirectRunner在Google Cloud Dataflow上运行能够成功运行的管道。当我执行此Maven命令时：

mvn compile exec:java \
    -Dexec.mainClass=com.example.Pipeline \
    -Dexec.args="--project=project-name \
    --stagingLocation=gs://bucket-name/staging/ \
    ... custom arguments ...
    --runner=DataflowRunner"

我收到以下错误：

No Runner was specified and the DirectRunner was not found on the classpath.
[ERROR] Specify a runner by either:
[ERROR]     Explicitly specifying a runner by providing the 'runner' property
[ERROR]     Adding the DirectRunner to the classpath
[ERROR]     Calling 'PipelineOptions.setRunner(PipelineRunner)' directly

我有意DirectRunner从我的收藏夹中删除pom.xml并添加了以下内容：

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
    <version>2.0.0</version>
    <scope>runtime</scope>
</dependency>

我继续删除了<scope>标签，然后调用了options.setRunner(DataflowRunner.class)，但没有帮助。PipelineOptions从扩展我自己的界面DataflowPipelineOptions …

我从事Apache Beam已有两天了。我想快速迭代正在使用的应用程序，并确保正在构建的管道没有错误。在spark中，我们可以使用sc.parallelise，当我们执行某些操作时，我们可以获得可以检查的值。

同样，当我阅读有关Apache Beam的内容时，我发现我们可以PCollection使用以下语法创建一个并使用它

with beam.Pipeline() as pipeline:
    lines = pipeline | beam.Create(["this is test", "this is another test"])
    word_count = (lines 
                  | "Word" >> beam.ParDo(lambda line: line.split(" "))
                  | "Pair of One" >> beam.Map(lambda w: (w, 1))
                  | "Group" >> beam.GroupByKey()
                  | "Count" >> beam.Map(lambda (w, o): (w, sum(o))))
    result = pipeline.run()

我实际上想将结果打印到控制台。但是我找不到关于它的任何文档。

有没有一种方法可以将结果打印到控制台，而不是每次都将其保存到文件中？

我正在使用Java Beam SDK进行数据流作业,而com.google.api.services.dataflow.model.Job类则提供有关特定作业的详细信息.但是,它不提供任何方法/属性来获取数据流步骤信息,例如添加元素,估计大小等.

下面是我用来获取工作信息的代码,

PipelineResult result = p.run();        
String jobId = ((DataflowPipelineJob) result).getJobId();
DataflowClient client = DataflowClient.create(options);
Job job = client.getJob(jobId);

我在找类似的东西,

job.getSteps("step name").getElementsAdded();
job.getSteps("step name").getEstimatedSize();

提前致谢.

我在GCS或其他受支持的文件系统上有一个目录,外部进程正在向该文件系统写入新文件.

我想写一个Apache Beam流管道,它不断地在这个目录中查看新文件,并在每个新文件到达时读取和处理它们.这可能吗？

我的工作流程:KAFKA - >数据流 - > BigQuery

鉴于在我的情况下低延迟并不重要,我使用FILE_LOADS来降低成本.我正在使用带有DynamicDestination的BigQueryIO.Write(每小时一个新表,当前小时作为后缀).

这个BigQueryIO.Write配置如下:

.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withMethod(Method.FILE_LOADS)
.withTriggeringFrequency(triggeringFrequency)
.withNumFileShards(100)

第一个表已成功创建并写入.但是,从未创建以下表格,我得到以下表格:

(99e5cd8c66414e7a): java.lang.RuntimeException: Failed to create load job with id prefix 5047f71312a94bf3a42ee5d67feede75_5295fbf25e1a7534f85e25dcaa9f4986_00001_00023, reached max retries: 3, last failed load job: {
  "configuration" : {
    "load" : {
      "createDisposition" : "CREATE_NEVER",
      "destinationTable" : {
        "datasetId" : "dev_mydataset",
        "projectId" : "myproject-id",
        "tableId" : "mytable_20180302_16"
      },

对于第一个表中的createDisposition会使用的CREATE_IF_NEEDED符合规定,但随后此参数不考虑和CREATE_NEVER默认使用.

我还在JIRA上创建了这个问题.

有没有办法ReadFromText在Python中使用转换读取多行csv文件？我有一个文件,其中包含一行,我试图让Apache Beam将输入读作一行,但无法让它工作.

def print_each_line(line):
    print line

path = './input/testfile.csv'
# Here are the contents of testfile.csv
# foo,bar,"blah blah
# more blah blah",baz

p = apache_beam.Pipeline()

(p
 | 'ReadFromFile' >> apache_beam.io.ReadFromText(path)
 | 'PrintEachLine' >> apache_beam.FlatMap(lambda line: print_each_line(line))
 )

# Here is the output:
# foo,bar,"blah blah
# more blah blah",baz

上面的代码将输入解析为两行,即使多行csv文件的标准是将多行元素包装在双引号内.

我目前正在从事ETL Dataflow作业（使用Apache Beam Python SDK），该作业从CloudSQL查询数据（带有psycopg2和自定义ParDo）并将其写入BigQuery。我的目标是创建一个数据流模板，该模板可以使用Cron作业从AppEngine开始。

我有一个使用DirectRunner在本地工作的版本。为此，我使用CloudSQL（Postgres）代理客户端，以便可以连接到127.0.0.1上的数据库。

当将DataflowRunner与自定义命令一起使用来在setup.py脚本中启动代理时，该作业将不会执行。它坚持重复此日志消息：

Setting node annotation to enable volume controller attach/detach

我的setup.py的一部分看起来如下：

CUSTOM_COMMANDS = [
['echo', 'Custom command worked!'],
['wget', 'https://dl.google.com/cloudsql/cloud_sql_proxy.linux.amd64', '-O', 'cloud_sql_proxy'],
['echo', 'Proxy downloaded'],
['chmod', '+x', 'cloud_sql_proxy']]

class CustomCommands(setuptools.Command):
  """A setuptools Command class able to run arbitrary commands."""

  def initialize_options(self):
    pass

  def finalize_options(self):
    pass

  def RunCustomCommand(self, command_list):
    print('Running command: %s' % command_list)
    logging.info("Running custom commands")
    p = subprocess.Popen(
        command_list,
        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    # Can use communicate(input='y\n'.encode()) if the command run …

我正在阅读一篇有关由某些Dataflow源和接收器实施的一次精确处理的文章，但在理解BigQuery接收器上的示例时遇到了麻烦。从文章

生成随机UUID是非确定性操作，因此在插入BigQuery之前，我们必须添加一个重新排列。完成此操作后，Cloud Dataflow进行的任何重试将始终使用改组后的相同UUID。重复插入BigQuery的尝试将始终具有相同的插入ID，因此BigQuery可以对其进行过滤

// Apply a unique identifier to each record
c
 .apply(new DoFn<> {
  @ProcessElement
  public void processElement(ProcessContext context) {
   String uniqueId = UUID.randomUUID().toString();
   context.output(KV.of(ThreadLocalRandom.current().nextInt(0, 50),
                                     new RecordWithId(context.element(), uniqueId)));
 }
})
// Reshuffle the data so that the applied identifiers are stable and will not change.
.apply(Reshuffle.of<Integer, RecordWithId>of())
// Stream records into BigQuery with unique ids for deduplication.
.apply(ParDo.of(new DoFn<..> {
   @ProcessElement
   public void processElement(ProcessContext context) {
     insertIntoBigQuery(context.element().record(), context.element.id());
   }
 }); …

SDK：适用于Go 0.5.0的Apache Beam SDK

数周以来，我们的Golang作业在Google Cloud Data流上运行良好。我们尚未对作业本身进行任何更新，并且SDK版本似乎与以前相同。昨晚失败了，我不确定为什么。到达1小时的时间限制，由于没有工人活动，该作业被取消。

查看Stackdriver日志，我唯一能看到的就是重复出现的错误 Error syncing pod...failed to "StartContainer" for "sdk" with CrashLoopBackOff

似乎是由于某种原因未能同步pod（？），因此需要等待5分钟才能重试。

任何人都可以阐明造成这种情况的原因以及我们如何找到更多信息或诊断问题的原因吗？

注意：我检查了Google Cloud Data flow的状态，该服务似乎没有任何中断。

我正在努力获取基于事件时间的触发器来触发我的apache光束管道，但是似乎确实能够触发具有处理时间的窗口触发。

我的管道相当基本：

1. 我收到了一批数据点，其中包括从pubsub读取的毫秒级时间戳，时间戳比最早批处理的数据点稍早。批量处理数据可以减少客户端的工作量和发布费用。

2. 我提取第二级时间戳并将时间戳应用于各个数据点

3. 我对数据进行窗口处理，并避免使用全局窗口。

4. 我将数据按秒进行分组，以供日后按流数据进行分类。

5. 最终，我最终在分类的几秒钟上使用了滑动窗口，以每秒有条件地将两条消息之一发送到pubsub。

我的问题似乎在步骤3中。

我试图在第3阶段使用最终将在第5阶段使用的相同的窗口化策略，以对分类的秒数进行滑动平均值计算。

我已经尝试过使用withTimestampCombiner（TimestampCombiner.EARLIEST）选项，但这似乎无法解决。

我已经读过有关事件时间使用的.withEarlyFirings方法，但这似乎可以模仿我现有的解决方法。理想情况下，我将能够依靠水印通过窗口的结尾并包含延迟触发。

// De-Batching The Pubsub Message

  static public class UnpackDataPoints extends DoFn<String,String>{
    @ProcessElement
        public  void processElement(@Element String c, OutputReceiver<String> out) {
            JsonArray packedData = new JsonParser().parse(c).getAsJsonArray();
            DateTimeFormatter dtf = DateTimeFormat.forPattern("EEE dd MMM YYYY HH:mm:ss:SSS zzz");
            for (JsonElement acDataPoint: packedData){
                String hereData = acDataPoint.toString();
                DateTime date = dtf.parseDateTime(acDataPoint.getAsJsonObject().get("Timestamp").getAsString());
                Instant eventTimeStamp = date.toInstant();
                out.outputWithTimestamp(hereData,eventTimeStamp);
            }
        }
        }

// Extracting The Second
 static public class ExtractTimeStamp extends DoFn<String,KV<String,String>> {
    @ProcessElement
        public void processElement(ProcessContext …