标签: dataflow

我以前从未使用过 TPL，所以我想知道是否可以用它来完成：我的应用程序从很多帧创建了一个 gif 图像动画文件。我从一个表示 gif 文件帧的 Bitmap 列表开始，需要对每一帧执行以下操作：

1. 在框架上绘制一些文本/位图
2. 裁剪框架
3. 调整框架大小
4. 将图像减少到 256 色

显然，这个过程可以对列表中的所有帧并行完成，但对于每个帧，步骤的顺序需要相同。之后，我需要将所有帧写入 gif 文件。因此，所有帧都需要按照它们在原始列表中的相同顺序接收。最重要的是，这个过程可以在第一帧准备好时开始，不需要等到所有帧都处理完。

所以情况就是这样。TPL Dataflow 适合这个吗？如果是的话，谁能给我一个关于如何设计 tpl 块结构以反映上述过程的正确方向的提示？与我发现的一些样本相比，这对我来说似乎相当复杂。

我很抱歉提出关于这个一般问题的另一个问题，但我在 SO 上发现的问题似乎都与我的问题密切相关。

我有一个现有的、工作的数据流管道，它接受对象KV<Long, Iterable<TableRow>>并输出TableRow对象。这段代码在我们的生产环境中，运行没有问题。然而，我现在正在尝试使用直接运行器实现单元测试来测试此管道，但是单元测试在到达线路时失败

LinkedHashMap<String, Object> evt = (LinkedHashMap<String, Object>) row.get(Schema.EVT);

在管道中，抛出错误消息：

java.lang.ClassCastException：com.google.gson.internal.LinkedTreeMap 无法转换为 java.util.LinkedHashMap

现有数据流代码的简化版本如下所示：

public static class Process extends DoFn<KV<Long, Iterable<TableRow>>, TableRow> {

    /* private variables */

    /* constructor */

    /* private functions */

    @ProcessElement
    public void processElement(ProcessContext c) throws InterruptedException, ParseException {       

      EventProcessor eventProc = new EventProcessor();
      Processor.WorkItem workItem = new Processor.WorkItem();
      Iterator<TableRow> it = c.element().getValue().iterator();

      // process all TableRows having the same id
      while (it.hasNext()) {
        TableRow item = it.next();

        if …

我需要通过 Dataflow 和 Beam 运行从 GCS 到 BigQuery 的批处理作业。我的所有文件都是具有相同架构的 avro。我创建了一个数据流 java 应用程序，它在较小的数据集（~1gb，大约 5 个文件）上成功。但是当我尝试在更大的数据集（>500gb，>1000 个文件）上运行它时，我收到一条错误消息

java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.lang.RuntimeException: Failed to create load job with id prefix 1b83679a4f5d48c5b45ff20b2b822728_6e48345728d4da6cb51353f0dc550c1b_00001_00000, reached max retries: 3, last failed load job: ...

3 次重试后，它终止于：

Workflow failed. Causes: S57....... A work item was attempted 4 times without success....

这一步是对 BigQuery 的加载。

堆栈驱动程序说处理卡在步骤....10m00s...和

Request failed with code 409, performed 0 retries due to IOExceptions, performed 0 retries due to unsuccessful status codes.....

我查找了 409 错误代码，指出我可能有一个现有的作业、数据集或表。我已经删除了所有表并重新运行了应用程序，但它仍然显示相同的错误消息。

我目前仅限于 65 个工人，我让他们使用 …

我有一个相当简单的生产者-消费者模式，其中（简化）我有两个生产者，他们生产由一个消费者消费的输出。

为此，我使用 System.Threading.Tasks.Dataflow.BufferBlock<T>

一个BufferBlock对象被创建。一个Consumer是听这个BufferBlock，并处理任何接收到的输入。

send data to the同时有两个“生产者BufferBlock”

简化：

BufferBlock<int> bufferBlock = new BufferBlock<int>();

async Task Consume()
{
    while(await bufferBlock.OutputAvailable())
    {
         int dataToProcess = await outputAvailable.ReceiveAsync();
         Process(dataToProcess);
    }
}

async Task Produce1()
{
    IEnumerable<int> numbersToProcess = ...;
    foreach (int numberToProcess in numbersToProcess)
    {
         await bufferBlock.SendAsync(numberToProcess);
         // ignore result for this example
    }
}

async Task Produce2()
{
    IEnumerable<int> numbersToProcess = ...;
    foreach (int numberToProcess in numbersToProcess)
    {
         await bufferBlock.SendAsync(numberToProcess);
         // ignore result for …

我正在尝试使用此示例从 DataFlow 的 GCP Pub/Sub 检索数据。

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;

import avro.shaded.com.google.common.collect.Lists;
import com.google.auth.oauth2.GoogleCredentials;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.transforms.windowing.SlidingWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.joda.time.Duration;

import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;

public class StreamDemoConsumer {

    public static interface MyOptions extends DataflowPipelineOptions {
        @Description("Output BigQuery table <project_id>:<dataset_id>.<table_id>")
        @Default.String("coexon-seoul-dev:ledger_data_set.ledger_data2")
        String getOutput();

        void setOutput(String s);

        @Description("Input topic")
        @Default.String("projects/coexon-seoul-dev/topics/trading") …

我正在使用 python 中的 Apache Beam（在 GCP Dataflow 上）做一个简单的管道，从 PubSub 读取并在 Big Query 上写入，但无法处理管道上的异常以创建替代流。

在一个简单的 WriteToBigQuery 示例中：

output = json_output | 'Write to BigQuery' >> beam.io.WriteToBigQuery('some-project:dataset.table_name')

我试图把它放在一个try/except代码中，但它不起作用，因为当它失败时，异常似乎被抛出到我的 python 执行之外的 Java 层上：

INFO:root:2019-01-29T15:49:46.516Z: JOB_MESSAGE_ERROR: java.util.concurrent.ExecutionException: java.lang.RuntimeException: Error received from SDK harness for instruction -87: Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 135, in _execute
    response = task()
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 170, in <lambda>
    self._execute(lambda: worker.do_instruction(work), work)
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 221, in do_instruction
    request.instruction_id)
...
...
...
    self.signature.finish_bundle_method.method_value())
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/io/gcp/bigquery.py", …

我正在尝试遵循缓慢变化的查找缓存的设计模式（https://cloud.google.com/blog/products/gcp/guide-to-common-cloud-dataflow-use-case-patterns-part-1）用于在 DataFlow 上使用适用于 Apache Beam 的 Python SDK 的流式传输管道。

我们的查找缓存参考表位于 BigQuery 中，我们能够读取它并将其作为 ParDo 操作的侧输入传递，但无论我们如何设置触发器/窗口，它都不会刷新。

class FilterAlertDoFn(beam.DoFn):\n  def process(self, element, alertlist):\n\n    print len(alertlist)\n    print alertlist\n\n    \xe2\x80\xa6  # function logic\n

alert_input = (p | beam.io.Read(beam.io.BigQuerySource(query=ALERT_QUERY))\n                        | \xe2\x80\x98alert_side_input\xe2\x80\x99 >> beam.WindowInto(\n                            beam.window.GlobalWindows(),\n                            trigger=trigger.RepeatedlyTrigger(trigger.AfterWatermark(\n                                late=trigger.AfterCount(1)\n                            )),\n                            accumulation_mode=trigger.AccumulationMode.ACCUMULATING\n                          )\n                       | beam.Map(lambda elem: elem[\xe2\x80\x98SOMEKEY\xe2\x80\x99])\n)\n\n...\n\n\nmain_input | \xe2\x80\x98alerts\xe2\x80\x99 >> beam.ParDo(FilterAlertDoFn(), beam.pvalue.AsList(alert_input))\n

基于此处的 I/O 页面 ( https://beam.apache.org/documentation/io/built-in/ )，它表示 Python SDK 仅支持 BigQuery Sink 的流式传输，这是否意味着 BQ 读取是有界源那么在这个方法中能不能刷新\xe2\x80\x99呢？

尝试在源上设置非全局窗口会导致侧输入中出现空 PCollection。

更新 …

我有一个 Azure Data Lake Store gen1 (ADLS-1) 和一个带有数据流 (DF) 的 Azure 数据工厂 (ADF) (V2)。当我在 ADF 中创建新的 DF 并在源和/或接收器节点中选择 ADLS-1 中的数据集时，我收到以下验证错误（在 DF 中）：

source1 AzureDataLakeStore does not support MSI authentication in Data Flow.

这是否意味着我无法将 DF 与 ADLS-1 一起使用，或者这是某种身份验证问题？

我尝试过的事情列表：

• 我已在 ADLS-1 的访问控制 (IAM) 中为 ADF 资源授予所有者角色
• 我已向 ADF 资源授予数据集 ADLS-1 文件夹中的所有（读、写等）权限
• 我可以在 ADF 管道中将数据从 ADLS-1 复制到 ADLS-1（因此在 DF 之外）
• 我可以在 DF 的源节点和接收器节点中为来自 ADLS-2 (gen 2) 的数据集选择数据集（所以这里我没有收到错误）
• 我可以创建一个管道，首先将数据集从 ADLS-1 复制到 ADLS-2，然后使用 DF 对其进行处理（然后将其复制回来）。这个解决方法非常乏味，而且我（目前）还没有生产 ADLS-2。
• 这里说ADLS-1 支持的功能包括映射数据流 (DF)。

如果有人知道将 DF 与 ADLS-1 …

我有一个关于 的背景下的侧面输入和广播的一般性问题Apache Beam。期间计算所需的任何其他变量、列表、映射是否processElement需要作为辅助输入传递？如果它们作为普通构造函数参数传递可以吗DoFn？例如，如果我有一些固定（未计算）值变量（常量，如开始日期、结束日期），我想在processElement. PCollectionView现在，我可以分别从每个变量中创建单例，并将它们DoFn作为侧面输入传递给构造函数。但是，我可以不这样做，而只是将每个常量作为普通构造函数参数传递给 吗DoFn？我在这里错过了一些微妙的东西吗？

就代码而言，我什么时候应该这样做：

public static class MyFilter extends DoFn<KV<String, Iterable<MyData>> {
  // these are singleton views
  private final PCollectionView<LocalDateTime> dateStartView;
  private final PCollectionView<LocalDateTime> dateEndView;

  public MyFilter(PCollectionView<LocalDateTime> dateStartView,
                       PCollectionView<LocalDateTime> dateEndView){

      this.dateStartView = dateStartView;
      this.dateEndView = dateEndView;
  }

  @ProcessElement
  public void processElement(ProcessContext c) throws Exception{
  // extract date values from the singleton views here and use them

与以下相反：

public static class MyFilter extends DoFn<KV<String, Iterable<MyData>> { …

我正在阅读有关 Apache BEAM 中的SCHEMAS 的文档，但我无法理解它的目的是什么、如何、为什么或在哪些情况下我应该使用它们。使用模式或使用扩展 Serialized 接口的类有什么区别？

该文档有一个示例：

@DefaultSchema(JavaFieldSchema.class)
public class TransactionPojo {
  public String bank;
  public double purchaseAmount;
}

PCollection<TransactionPojos> transactionPojos = readTransactionsAsPojo();

但它没有解释readTransactionsAsPojo函数是如何构建的。我认为对此有很多遗漏的解释。