请参阅下面的代码片段，我想["metric1", "metric2"]成为 RunTask.process 的输入。但是，它分别使用“metric1”和“metric2”运行了两次

def run():
  
  pipeline_options = PipelineOptions(pipeline_args)
  pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
  p = beam.Pipeline(options=pipeline_options)

  root = p | 'Get source' >> beam.Create([
      "source_name" # maybe ["source_name"] makes more sense since my process function takes an array as an input?
  ])

  metric1 = root | "compute1" >> beam.ParDo(RunLongCompute(myarg="1")) #let's say it returns ["metic1"]
  metric2 = root | "compute2" >> beam.ParDo(RunLongCompute(myarg="2")) #let's say it returns ["metic2"]

  metric3 = (metric1, metric2) | beam.Flatten() | beam.ParDo(RunTask()) # I want ["metric1", "metric2"] …

太长了；

如何使用“environment_type”= EXTERNAL 或 PROCESS 配置 Apache Beam 管道选项？

描述

目前，我们在 Kubernetes 中有一个独立的 Spark 集群，按照这个解决方案（和设置），我们启动了一个 Beam 管道，在需要联合运行 Python SDK 的 Spark Worker 上创建一个嵌入式 Spark 作业服务器。Apache Beam 允许以4 种不同的方式运行 python SDK ：

• “DOCKER” - 默认且在 Kubernetes 集群内不可能（将使用“容器内的容器”）
• “LOOPBACK” - 仅用于测试，不可能有超过 1 个工作 Pod
• “EXTERNAL” - 理想的设置，“只需”创建一个 sidecar 容器，与 Spark Worker 在同一个 Pod 中运行
• “PROCESS” - 在 Spark Worker 中执行一个进程，虽然不理想，但也可以。

发展

1. 使用“外部”- 在同一 pod 上使用 python sdk 实现 Spark Worker：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: spark-worker
  labels: …

在进行一些处理并按键分组后，我得到了如下所示的数据集。我现在需要对每一行数据进行一些处理以获得下面的输出。我尝试过平面地图，它真的很慢，因为“值”列表的长度可以是任意长的。我想我可以将每一行分成单独的 pcollections，并行处理，然后将它们压平在一起。如何将每一行拆分为不同的 pcollection？如果这不可行，是否有其他方法可以加快计算速度？

输入

key, value
1    [A, B, B, B]
2    [A, B, B, B]
3    [A, B, B, B]
4    [A, B, B, B]
5    [A, B, B, B]

输出：

key, value
1    (A, 0)
1    (B, 1)
1    (B, 2)
1    (B, 3)
2    (A, 0)
2    (B, 1)
2    (B, 2)
2    (B, 3)
...

我想知道是否有可能/在 AWS 中实现 TFX 管道（在真实数据集上，具有 100+ GB 数据集，而不是具有小数据集的教程）有多容易？

对于编排，我可能会使用 Kubeflow。但我认为，主要问题是为 Apache Beam 设置适当的可扩展运行程序。我正在考虑使用 Apache Flink 来实现这一点。

有谁有做经验的吗？当您需要定期使用新数据训练模型时，您会如何在 AWS 中将 TF 投入生产，是从头开始编写管道还是使用某种工具？

我们在使用 Python 和 Java 编写的 GCP 数据流上运行 Beam 数据管道。一开始，我们有一些简单直接的 Python Beam 作业，效果非常好。因此，最近我们决定将更多的 Java Beam 作业转换为 Python Beam 作业。当我们有更复杂的工作时，尤其是需要在光束中开窗的工作时，我们注意到 python 工作比 java 工作明显慢，最终使用更多的 cpu 和内存并且成本更高。

一些示例 python 代码如下所示：

        step1 = (
        read_from_pub_sub
        | "MapKey" >> beam.Map(lambda elem: (elem.data[key], elem))
        | "WindowResults"
        >> beam.WindowInto(
            beam.window.SlidingWindows(360,90),
            allowed_lateness=args.allowed_lateness,
        )
        | "GroupById" >> beam.GroupByKey()

Java 代码如下：

 PCollection<DataStructure> step1 =
      message
          .apply(
              "MapKey",
              MapElements.into(
                      TypeDescriptors.kvs(
                          TypeDescriptors.strings(), TypeDescriptor.of(DataStructure.class)))
                  .via(event -> KV.of(event.key, event)))
          .apply(
              "WindowResults",
              Window.<KV<String, CustomInterval>>into(
                      SlidingWindows.of(Duration.standardSeconds(360))
                          .every(Duration.standardSeconds(90)))
                  .withAllowedLateness(Duration.standardSeconds(this.allowedLateness))
                  .discardingFiredPanes())
          .apply("GroupById", GroupByKey.<String, DataStructure>create())

我们注意到 Python 使用的 CPU …

我正在尝试设置我的开发环境.我一直在使用pubsub模拟器进行开发和测试,而不是在生产中使用google cloud pubsub.为此,我设置了以下环境变量:

export PUBSUB_EMULATOR_HOST=localhost:8586

这适用于python google pubsub库但是当我切换到使用java apache beam for google dataflow时,管道仍指向生产google pubsub.管道上是否有设置,环境变量或方法需要设置,以便管道读取本地pubsub模拟器？

我需要从GCS存储桶中读取文件.我知道我将不得不使用GCS API /客户端库,但我找不到任何与之相关的示例.

我一直在参考GCS文档中的这个链接: GCS客户端库.但真的不能成功.如果有人能提供一个真正有用的例子.谢谢.

我有一个Dataflow作业要写入BigQuery.它适用于非嵌套模式,但嵌套模式失败.

这是我的Dataflow管道:

pipeline_options = PipelineOptions()
  p = beam.Pipeline(options=pipeline_options)

  wordcount_options = pipeline_options.view_as(WordcountTemplatedOptions)

  schema = 'url: STRING,' \
           'ua: STRING,' \
           'method: STRING,' \
           'man: RECORD,' \
           'man.ip: RECORD,' \
           'man.ip.cc: STRING,' \
           'man.ip.city: STRING,' \
           'man.ip.as: INTEGER,' \
           'man.ip.country: STRING,' \
           'man.res: RECORD,' \
           'man.res.ip_dom: STRING'

  first = p | 'read' >> ReadFromText(wordcount_options.input)
  second = (first
            | 'process' >> (beam.ParDo(processFunction()))
            | 'write' >> beam.io.WriteToBigQuery(
              'myBucket:tableFolder.test_table',
              schema=schema)
  )

我使用以下Schema创建了BigQuery Table:

[
  {
    "mode": "NULLABLE",
    "name": "url",
    "type": "STRING"
  },
  {
    "mode": "NULLABLE", …

我正在使用我的Mac上的DirectRunner测试我的Dataflow管道并得到了很多像这样的"警告"消息,我可能知道如何摆脱它们,因为它太多了,我甚至看不到我的调试消息.

谢谢

Apr 05, 2018 2:14:48 PM org.apache.beam.sdk.util.MutationDetectors$CodedValueMutationDetector verifyUnmodifiedThrowingCheckedExceptions
WARNING: Coder of type class org.apache.beam.sdk.coders.SerializableCoder has a #structuralValue method which does not return true when the encoding of the elements is equal. 
Element com.apigee.analytics.platform.core.service.schema.EventRow@4a590d0b

我有一个用例，需要将70 TB的数据从DynamoDB迁移到BigTable和Spanner。具有单个索引的表将进入BigTable，否则将进入Spanner。

通过将数据导出到S3-> GCS-> Spanner / BigTable，可以轻松处理历史负载。但是具有挑战性的部分是要处理DynamoDB上同时发生的增量流负载。DynamoDB中有300个表。

如何以最好的方式处理这件事？有人做过吗？