Apache光束/数据流中跨多个文件的流水线代码

Apache Beam是一个开源的分布式数据处理框架，它可以在多个文件之间进行流水线代码的处理。它提供了一种统一的编程模型，可以处理批处理和流处理数据，并且可以在各种分布式处理引擎上运行，如Apache Flink、Apache Spark和Google Cloud Dataflow。

Apache Beam的主要特点包括：

统一的编程模型：Apache Beam提供了一种统一的编程模型，可以同时处理批处理和流处理数据。开发人员可以使用相同的代码逻辑来处理不同类型的数据。
分布式处理：Apache Beam可以在分布式处理引擎上运行，充分利用集群资源进行高效的数据处理。它可以自动进行任务并行化和资源管理，以提高处理性能。
可扩展性：Apache Beam可以处理大规模的数据集，并且可以根据需求进行水平扩展。它可以自动处理数据的分片和分发，以实现高吞吐量和低延迟的数据处理。
容错性：Apache Beam具有容错机制，可以处理节点故障和数据丢失。它可以自动进行故障恢复和数据重放，以确保数据处理的可靠性和一致性。
生态系统支持：Apache Beam拥有丰富的生态系统，提供了各种数据处理和转换操作，如过滤、映射、聚合、连接等。开发人员可以根据自己的需求选择适合的操作来构建数据处理流水线。

Apache Beam适用于各种数据处理场景，包括实时数据分析、批处理任务、ETL（抽取、转换、加载）流程等。它可以处理各种类型的数据，如结构化数据、半结构化数据和非结构化数据。

对于Apache Beam的具体使用和相关产品推荐，可以参考腾讯云的数据计算服务TencentDB for Apache Beam（https://cloud.tencent.com/product/beam）和数据流计算服务Tencent Cloud StreamCompute（https://cloud.tencent.com/product/sc）。

总结：Apache Beam是一个开源的分布式数据处理框架，可以在多个文件之间进行流水线代码的处理。它具有统一的编程模型、分布式处理、可扩展性、容错性和丰富的生态系统支持。适用于各种数据处理场景，如实时数据分析、批处理任务和ETL流程。腾讯云提供了相关的数据计算和数据流计算服务，可以满足用户的需求。

Apache光束/数据流中跨多个文件的流水线代码

python、google-cloud-dataflow、apache-beam

经过长时间的搜索，我还没有找到一个跨多个文件的数据流/光束管道的示例。文档做了 (在“多个文件依赖项”部分下)，但是他们给出的Juliaset示例实际上有一个文件(以及调用它的Beam )。基于Juliaset示例，我需要一个类似的文件结构：juliaset/juliase

浏览 0提问于2017-12-16得票数 4

1回答

如何在数据流管道中实现PubSubIO中的流量控制设置

google-cloud-platform、google-bigquery、google-cloud-dataflow、apache-beam、google-cloud-pubsub

在我们的应用程序中，我们在数据流管道中使用PubsubIO从PubSub读取数据。下面是代码。from bigquery table", ParDo.of(new ReadRawdataFromBiqueryTable())); 但是，当我们在此管道中附加BigQuery读取时，由于BigQuery读取速度较慢，因为我们在ParDo中执行此操作，似乎在PubSub订阅服务器中实现了一些默认的流控制

浏览 12提问于2020-03-25得票数 1

1回答

为什么在Scio中你更喜欢聚合而不是groupByKey？

scala、apache-beam、dataflow、spotify-scio

记住，减少操作必须是关联的和可交换的。” 尤其是为什么人们更喜欢聚合而不是groupByKey呢？

浏览 1提问于2018-05-11得票数 1

回答已采纳

2回答

如果我们的类路径中有2.2.0，那么我的管道使用哪个版本的SDK？

google-cloud-dataflow

beam-sdks-java-core的最新版本是2.3.0。然而，如果我的流水线代码使用beam-sdks-java-core 2.2.0，那么我的流水线在Dataflow中运行时会使用2.2.0还是2.3.0？

浏览 1提问于2018-03-10得票数 0

6回答

Google Cloud Dataflow和Google Cloud Dataproc之间的区别是什么？

google-cloud-platform、google-cloud-dataflow、google-cloud-dataproc

看看谷歌的云服务，DataProc似乎也可以做同样的事情。为什么谷歌同时提供这两种服务？

浏览 1提问于2017-09-27得票数 66

1回答

Apache / PubSub处理文件前的时间延迟

google-cloud-platform、google-cloud-dataflow、apache-beam、google-cloud-pubsub

我需要延迟处理或发布文件名(文件)。我在找最好的选择。1)延迟发布消息，以便

浏览 0提问于2018-12-21得票数 0

1回答

Simulink并行运行块？

matlab、parallel-processing、simulink

假设我在Simulink中有一个很大的模型，我们称它为模型A，现在考虑模型A的一个很小的子集，称之为模型B。当B模型计算某件事情时，这些结果被发送到模型A的其他部分，它们作为模型A的函数做某些事情。然而，模型B可能需要很长的时间来计算--然而，这对于模型A的其他块来说并不是一个问题，只要模型B完成计算，它们就很乐意从模型B接收数据。在Simulink中可以这样做吗？

浏览 0提问于2016-06-15得票数 1

1回答

将多个Apache光束管道打包到一个jar文件中

java、maven、executable-jar、apache-beam

我正在处理一个项目，其中包含许多用Java编写的光束管道，需要将其打包为jar文件，以便从我们的作业调度器执行。我尝试使用构建配置文件为每个main创建一个jar，但这看起来很混乱，而且我遇到了依赖冲突的问题(当它不使用时，它仍然在寻找所需的区域选项)。我也只是寻找一个不断增长的光束代码库的整体可持续的项目结构建议。打包要按计划执行的管道的最佳

浏览 18提问于2019-01-05得票数 0

2回答

Flink将Apache光束捆绑在一起，不显示单独的任务块

apache-flink、apache-beam

我有一个运行在Flink runner上的光束管道，如下所示： ? 光束中的整个管道都被捆绑到这个区块中。为什么Flink用这种方式来说明它？ ? 当然，管道步骤中的不同pardo应该由不同的块和工作者来表示(就像在Google的数据流中一样)。是否需要在Flink运行器中设置某些内容以显示更细粒度的任务，或者Flink UI是否仅为原生Flink<

浏览 24提问于2021-07-05得票数 1

1回答

GCP Cloud Functions是处理大型.csv文件的正确工具吗？

google-cloud-platform、google-cloud-dataflow、google-cloud-functions

我的目标是允许用户将相对较大的数据集导入MongoDB (托管在mongo云上)。Api服务器托管在容器引擎中，它没有任何其他负载如此大的操作(只有简单的CRUD操作)。用户将上传.csv文件(高达100mb)到GCP Bucket，从那里我需要验证和插入的csv文件到MongoDB的每一行，然后做一些计算的基础上导入的数据。这是否适合GCP云函数？我有点担心540秒的执行时间限制

浏览 1提问于2017-11-09得票数 1

1回答

如何处理apache beam管道中的异常？

java、google-cloud-dataflow、apache-beam、apache-beam-io

我正在用java构建一个apache光束管道，它做了一系列的事情，比如读取文件，创建审计，并将其加载到bigquery。如果我的管道在任何步骤失败，我希望将文件移动到其他文件夹。所以基本上，如果我的流水线在任何步骤由于任何原因而失败，我想在下面的if条件下处理它。目前，如果流水线成功运行，那么它将进入else部分并完成打印，但如果流水线失败，它将显示退出代码1，但不会打印失败。有没

浏览 25提问于2021-03-17得票数 0

1回答

在Flink集群上运行的Apache光束管道失败

apache-flink、apache-beam

我尝试在本地部署的Flink Docker集群上部署一个Apache光束管道。:108)Pipeline p= Pipeline.create(options); 并且在定义<em

浏览 80提问于2020-08-20得票数 1

1回答

在数据流模板中调用waitUntilFinish()之后，您可以运行代码吗？

java、google-cloud-dataflow、apache-beam

我有一个批处理Apache梁作业，它从GCS获取一个文件作为输入。我的目标是将文件移动到两个GCS存储桶中的一个，这取决于执行后管道的状态。如果流水线执行成功，则将文件移至存储桶A，否则，如果流水线在执行过程中出现任何未处理的异常，则将文件移至存储桶B。我正在使用Java的Apache Beam2.24.0版本，我需要创建一个Dataflow模板

浏览 21提问于2020-09-24得票数 0

1回答

向pcollection添加时间戳

apache-beam

我是一个非常新手，正在使用一个简单的文本文件批处理加载过程。我想为在BigQuery中插入记录添加一个时间戳。是否有为PCollection添加“插入日期”的首选模式？我已经看到了几种不同的方法，但我想知道是否有更好的模式或最佳实践？谢谢!

浏览 14提问于2020-09-02得票数 0

1回答

数据流状态持久化规范

google-cloud-platform、google-cloud-dataflow

我们正在考虑使用波束/数据流进行有状态处理。我们理解状态持久性是由Windmill/Persistent Disks 1管理的。这项持久化技术与其他替代技术(例如用于Flink的RocksDB )相比是否存在已知的局限性？谢谢！

浏览 4提问于2021-11-04得票数 0

1回答

有没有办法在GCP数据流使用apache beam完成工作后进行处理？

java、google-cloud-dataflow、apache-beam

在GCP数据流状态为done/completed后，是否可以进行后期处理。我有一个进程，dataflow从GCP存储中批量读取一个文件，并执行一些外部api调用进行转换，然后写回另一个文件。在所有批处理完成后，我需要做一些额外的处理。有没有办法做到这一点？我正在使用Apache光束和模板来运行GCP数据流。

浏览 2提问于2021-07-30得票数 0

3回答

Apache光束GCP在动态创建的目录中上传Avro

google-cloud-platform、pipeline、dataflow、apache-beam

我想在GCP中创建一个流式Apache光束管道，它从Google Pub/Sub读取数据并将其推送到GCS。我有一个可以从发布/订阅中读取数据的位。我的当前代码看起来就像这样(从GCP Apache beam模板中挑选出来的) pipeline.apply("Read PubSub Events", PubsubIO.readMessagesWithAttributes-04-28T09:0

浏览 25提问于2020-04-28得票数 1

回答已采纳

1回答

如何在apache beam管道中记录传入消息

google-cloud-dataflow、apache-beam、google-cloud-pubsub、apache-beam-io

我正在编写一个简单的apache光束流水线，从pubsub主题获取输入并将其存储到bigquery中。：logging.info(events)人们如何开发/调试这些流水线？

浏览 24提问于2019-07-06得票数 1

回答已采纳

1回答

在流式管道中组合多个端输入时，数据流失败

python、google-cloud-dataflow、apache-beam

我已经用Python SDK (Apache光束Python3.7SDK 2.19.0)构建了一个窗口数据流管道。beam.io.WriteToPubSub(topic=output_topic)这在本地运行良好(使用DirectRunner)，但在GCP (DataflowRunner)中运行时会失败我可以在数据流日志中看到一个java异常，其中包含： Caused by: org.apache.beam.sdk.coders.CoderException

浏览 12提问于2020-02-21得票数 0

1回答