标签: apache-beam

让我简化一下我的案子.我正在使用Apache Beam 0.6.0.我的最终处理结果是PCollection<KV<String, String>>.我想将值写入与其键对应的不同文件.

例如,假设结果包含

(key1, value1)
(key2, value2)
(key1, value3)
(key1, value4)

然后,我想写value1,value3并value4到key1.txt,并写入value4到key2.txt.

在我的情况下:

• 密钥集是在管道运行时确定的,而不是在构建管道时确定的.
• 密钥集可以非常小,但是对应于每个密钥的值的数量可能非常大.

有任何想法吗？

我们想要构建一个Cloud Dataflow Streaming流水线，该流水线从Pubsub接收事件，并对每个事件执行多个类似ETL的操作。这些操作之一是每个事件都有一个设备ID，需要将其转换为不同的值（我们称其为mapping -id），设备ID-> mapped-id的映射由外部服务通过REST API。同一设备ID可能会在多个事件中重复-因此，这些设备ID->映射的ID映射可以被缓存和重新使用。由于我们可能每秒通过管道最多处理每秒3M事件，因此需要尽可能减少对REST API的调用，并在实际需要调用时对其进行优化。

考虑到此设置，我有以下问题。

• 为了优化REST API调用，Dataflow是否提供任何内置的优化（如连接池），或者如果我们决定使用自己的此类机制，是否需要牢记任何限制/限制？

• 我们正在研究一些内存中缓存选项，以本地缓存映射，其中一些也由本地文件支持。因此，在不影响工作程序中常规Dataflow操作的情况下，这些缓存可以使用多少内存（占实例内存的一部分）是否有任何限制？如果我们使用文件支持的缓存，那么每个工作程序上是否都有可以由应用程序本身安全地用于构建这些文件的路径？

• 唯一设备ID的数量可能约为数百万个-因此并非所有这些映射都可以存储在每个实例中。因此，为了能够更好地利用本地缓存，我们需要在device-id和处理它们的工人之间获得某种亲和力。在进行此转换之前，我可以根据device-id进行分组。如果这样做，是否可以保证相同的设备ID或多或少由同一工作人员处理？如果有某种合理的保证，那么除了第一次调用就可以了，我大多数时候都不必使用外部REST API。还是有更好的方法来确保ID与工作人员之间的这种亲和力。

谢谢

在数据流流世界中。

我说的时候的理解：

Window.into(FixedWindows.of(Duration.standardHours(1)))
  .triggering(AfterProcessingTime.pastFirstElementInPane()
      .plusDelayOf(Duration.standardMinutes(15))

就是对于一个小时的固定窗口，触发器在看到第一个元素之后将等待或批处理这些元素。

但是当我说：

Window.into(FixedWindows.of(Duration.standardHours(1)))
  .triggering(AfterProcessingTime.pastFirstElementInPane()

它是从第一次看到第一个元素起就每次触发，还是隐式地对元素进行批处理？因为在每个元素上触发都会使系统过载。

尝试在Dataflow服务上运行管道时,我在命令行上指定了staging和temp buckets(在GCS中).当程序执行时,我在管道运行之前得到一个RuntimeException,其根本原因是我在路径中遗漏了一些东西.

Caused by: java.lang.RuntimeException: Failed to construct instance from factory method DataflowRunner#fromOptions(interface org.apache.beam.sdk.options.PipelineOptions) ... Caused by: java.lang.IllegalArgumentException: Missing object or bucket in path: 'gs://df-staging-bucket-57763/', did you mean: 'gs://some-bucket/df-staging-bucket-57763'?

gs://df-staging-bucket-57763/已存在于我的项目中,我可以访问它.我需要添加什么才能使其工作？

我正在尝试使用Beam管道,以便将SequenceMatcher函数应用于大量单词.我(希望)除了WriteToText部分之外已经找到了所有的东西.

我已经定义了一个自定义ParDo(在此称为ProcessDataDoFn),它接受main_input和side_input,处理它们并输出像这样的字典

{u'key': (u'string', float)}

我的管道非常简单

class ProcessDataDoFn(beam.DoFn):
    def process(self, element, side_input):

    ... Series of operations ...

    return output_dictionary

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

    # Main input
    main_input = p | 'ReadMainInput' >> beam.io.Read(
        beam.io.BigQuerySource(
            query=CUSTOM_SQL,
            use_standard_sql=True
        ))

    # Side input
    side_input = p | 'ReadSideInput' >> beam.io.Read(
        beam.io.BigQuerySource(
            project=PROJECT_ID,
            dataset=DATASET,
            table=TABLE
        ))

    output = (
        main_input
        | 'ProcessData' >> beam.ParDo(
            ProcessDataDoFn(),
            side_input=beam.pvalue.AsList(side_input))
        | 'WriteOutput' >> beam.io.WriteToText(GCS_BUCKET)
    )

现在的问题是,如果我像这样离开管道,它只输出output_dictionary的键.如果我将ProcessDataDoFn的返回值更改为json.dumps(ouput_dictionary),则Json写得正确但是像这样

{
'
k
e
y
'

:

[
'
s
t
r …

我正在使用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();

提前致谢.

我在数据存储写入速度方面遇到了巨大的性能问题.大多数情况下它保持在100个元素/秒以下.

当使用数据存储客户端(com.google.cloud:google-cloud-datastore)在本地计算机上标记写入速度并且并行运行批量写入时,我能够实现大约2600个元素/秒的速度.

我使用Java API建立了一个简单的Apache Beam管道.这是图表:

在没有数据存储节点的情况下运行时的速度:

这种方式要快得多.这一切都指向DatastoreV1.Write成为这个管道中的瓶颈 - 通过没有写节点的管道速度和DatastoreV1.Write与其他节点的挂起时间相比的挂起时间来判断.

方法我试图解决这个问题:

•增加初始工人的数量(尝试过1和10,没有明显的差异).数据存储在一段时间后(可能在前2个节点完成处理之后)将写入次数减少到1.基于此我怀疑DatastoreIO.v1().write()不会并行运行其worker.为什么呢？

•确保所有内容都在同一位置运行:GCP项目,数据流管道工作者和元数据,存储 - 所有都设置为us-central.这个建议在这里

•尝试使用不同的实体密钥生成策略(根据本文).目前使用这种方法:Key.Builder keyBuilder = DatastoreHelper.makeKey("someKind", UUID.randomUUID().toString());.我不完全确定这会产生足够均匀分布的密钥,但我猜即使它不是性能也不应该这么低？

请注意,我无法使用提供的Apache Beam和Google库而没有解决方法:由于依赖性问题,我不得不强迫google-api-client版本为1.22.0&Guava为23.0(例如参见https) ://github.com/GoogleCloudPlatform/DataflowJavaSDK/issues/607).

查看DatastoreV1.Write节点日志:

它每5周左右就推动500个实体的批量生产,这不是很快.

总的来说,它看起来像DatastoreIO.v1().write()速度很慢,而且它的工作者并没有并行运行.任何想法如何解决这个或可能是什么原因？

我正在使用Google Dataflow中的Beam KafkaIO源运行作业，找不到在作业重新启动期间持久保存偏移的简单方法（作业更新选项不够，我需要重新启动作业）

将Beam的KafkaIO与PubSubIO进行比较（或者更精确地说是比较PubsubCheckpoint与KafkaCheckpointMark），我可以看到，在KafkaIO中并未实现检查点持久性（KafkaCheckpointMark.finalizeCheckpoint方法为空），而在PubsubCheckpoint.finalizeCheckpoint中实现了它，它对PubSub进行了确认。

这是否意味着我无法以最小的努力可靠地管理作业重启时的Kafka偏移量？

我到目前为止考虑的选项：

1. 实现我自己的持久偏移逻辑-听起来很复杂，我在Scala中使用Beam尽管Scio。

2. 不执行任何操作，但这将导致作业重新启动时出现许多重复（主题的保留期为30天）。

3. 启用自动提交，但这会导致丢失消息，甚至更糟。

带命令的气流安装失败

sudo pip3 install apache-airflow[gcp_api]

昨天一切都很好.今天我看到以下错误:

Could not find a version that satisfies the requirement apache-beam[gcp]==2.3.0 (from google-cloud-dataflow->apache-airflow[gcp_api]) (from versions: 0.6.0, 2.0.0, 2.1.0, 2.1.1, 2.2.0)
No matching distribution found for apache-beam[gcp]==2.3.0 (from google-cloud-dataflow->apache-airflow[gcp_api])

有人可以帮我吗？

提前致谢

我注意到的一件事是,BigQueryIO.read().fromQuery()的性能比Apache Beam中的BigQueryIO.read().from()的性能要慢得多.为什么会这样？有没有办法改善它？