我以字典的形式将来自Google Cloud Pubsub的(无界的)数据导入PCollection。随着流数据的传入，我想通过在静态(有界的)查找表上通过键连接它来丰富它。这张表足够小，可以存放在内存中。 我目前有一个使用DirectRunner运行的有效解决方案，但是当我尝试在DataflowRunner上
浏览 7提问于2019-09-05得票数 4
基本上，我们希望根据特定列(而不是日期)的值，将一个大的(数十亿行) bigquery表拆分成大量(可以是100k左右)较小的表。我不知道如何在bigquery中高效地完成这项工作，所以我在考虑使用数据流。所以在这个操作之后，我们有了( PCollection，记录)的键。然后我们需要将PCollection写回bigquery表，表名可以是key_table。) | beam.G
浏览 9提问于2017-07-13得票数 0
我有一个用例，我曾经通过流(Kinesis)获取数据，并希望对其执行一些转换。在转换过程中，我需要查找MongoDB/DocumentDB/Redis，其中存储了我们的参考数据。我正在通过apache beam实现这个用例。
我需要使用这些数据库(MongoDB/DocumentDB/Redis)中的集合/表作为辅助输入，以便加载一次并从那里进行查找
浏览 0提问于2020-12-15得票数 1
1回答
我想知道我是否可以在Google Dataflow平台上直接缓存数据集(就像在Spark中缓存RDDs一样)。如果没有这样的功能，Dataflow如何在应用程序中挑选热门数据集，特别是当您有多个热门数据集，并且您希望根据数据集的重要性来确定缓存的优先级时？
浏览 1提问于2017-09-02得票数 2
在这个中，Apache建议在写入BigQuery时使用死信模式。此模式允许您用'FailedRows'标记从转换输出中获取未能写入的行。/runners/common.py", line 1198,
in apache_beam.runners.common.DoFnRunner.process File "apache_beam"apache</e
浏览 34提问于2022-09-03得票数 0
回答已采纳
我想知道使用Apache / Beam是否合适，就像内存中的数据库一样--我想不断地重新计算并实现一个基于边缘触发更新的数据的特定视图。这是流Flink / Beam的合适用例吗？我在一个不同的系统中与Beam一起工作，但只在批处理模式下工作，我认为如果我能够边缘触发它，这将是一个很好的工具。我要坚持的部分是，在批处理模式下，PCollections都是“完整的
浏览 2提问于2022-03-27得票数 1
回答已采纳
我有一个在pubSub中创建主题并将消息发布到该主题的程序。我还有一个自动的数据流作业(使用模板)，它将这些消息保存到我的BigQuery表中。我开始用python编写脚本，并做了很多尝试和错误来实现它，但令我沮丧的是，我无法实现它。代码如下所示：from apache_beam.io import WriteToText
TOPIC_PATH
浏览 3提问于2019-09-16得票数 0
1回答
在我的apache beam和数据流管道中，我做了一些需要全局合并操作的转换，比如min，max，自定义全局合并函数。pcollection中要处理的项的数量在20-40亿左右。我研究了一下代码，groupByKey试图给所有元素添加一个静态的空键，然后进行分组，这是不是意味着我们在混洗数据(特别是当我们只有一个键的时候)？有没有有效地做到这一点的方法？另一个我自己理解的问题是:beam/
浏览 3提问于2018-03-20得票数 1
有人知道在google-cloud-dataflow中使用文件模式匹配时如何获取文件名吗？
我是newbee来使用数据流。如何在使用文件模式匹配时获取文件名。
浏览 2提问于2015-05-01得票数 5
我的最后处理结果是PCollection<KV<String, String>>。我想要将值写入与它们的键对应的不同文件。就我而言：
有什么想法吗？
浏览 9提问于2017-04-08得票数 9
2回答
我以前在堆栈溢出()上看到过这个问题的答案，但是自从apache为python添加了可拆分的dofn功能之后，我就没有看到这个问题了。当将文件模式传递给gcs桶时，如何访问正在处理的当前文件的文件名？我想将文件名传递到转换函数中：
with beam.Pipeline(options=pipeline_options) as p:最后，当我转换json的每一行时，我要做的是将文件名传递到转换函数中(请参阅，然后使用文件名在另一个BQ<
浏览 0提问于2018-11-21得票数 8
回答已采纳
我一直在测试如何使用自创建的数据连接pub/sub数据。下面是主要的管道方法。)我试着分别在GroupByKey上做('Hello', ['sh 1','sh 1.1'])
('Hello_world
浏览 0提问于2021-07-16得票数 0
回答已采纳
读取gs:/存储桶上的csv文件，创建BigQuery表并追加数据。as beamfrom apache_beam.io import WriteToTextfrom apache_beam.metrics import Metrics
from apache_beam.me
浏览 4提问于2022-01-30得票数 1
回答已采纳
我在Apache Beam (在数据流中运行)中遇到过这样的情况，我已经基于创建了一个简单的有状态DoFn。上游窗口是全局的，更改它会影响下游聚合。我正在考虑在数据上存储我自己的时间戳，并使用计时器定期清理该表。这样做可取吗？
存储的数据是一些事件数据上的缓存键。缓存键告诉我，我需要查找此事件的过去事件数据，以更新当前事件。我不确定这在数据流<e
浏览 0提问于2021-02-18得票数 0
2回答
我正在尝试用Dataflow创建一个流管道，它从PubSub主题中读取消息，最终将它们写入BigQuery表中。我不想使用任何数据流模板。目前，我只想在从Google实例执行的Python3脚本中创建一个管道，以执行来自Pubsub的每条消息的加载和转换过程(解析其中包含的记录并添加一个新字段)，最终将结果写入BigQuery表。那么记录的结构(字典中的Python)和字段的数据类型就是BigQuery表所期望的。如
浏览 0提问于2019-11-14得票数 3
回答已采纳
我试图遵循这样的设计模式，使用来缓慢地更改流管道的查找缓存( Lookup，)。查找缓存的引用表位于BigQuery中，我们可以将它作为一个侧输入读入到ParDo操作中，但是不管我们如何设置触发器/窗口，它都不会刷新。，它说Python只支持BigQuery Sink的流，这是否意味着be读取是一个有界的源，因此不能在此方法中刷新？尝试在源上设置非全局窗口会导致侧输入中出现空PCo
浏览 0提问于2019-03-08得票数 5
我一直在Python 2.7.16上使用2.13.0 SDK测试Apache Beam，以流模式从Google Pub/Sub订阅中提取简单消息，并写入Google Big Query表。和建议应该将生成的服务KV(如id_label )作为attributes属性的一部分返回，但是它们似乎没有返回。
请注意，只有在使用数据流运行器时才支持id_label参数。--tem
浏览 2提问于2019-07-24得票数 2
我必须在python中编写一个Dataflow作业，它将从GCS读取两个不同的.csv文件，执行一个连接操作，对连接数据的结果执行转换，然后最后将其发送到BigQuery表？我对此非常陌生，我知道在经过大量的研发之后，我们可以从apache.beam完成所有的流水线操作，我终于找到了一个模板，但在给定的点上仍然有很多的混乱。如何从桶中读取模式(上面我发现它可以像这样读取，但我怀疑它是否能像上面那样读
浏览 1提问于2020-06-21得票数 4
我想在Google上读取来自MariaDB的数据，并通过DataFlow将数据写入BigQuery，但在DataFlowRunner上运行DataFlow程序时，我总是会遇到以下异常。address=(host=xxx.xxx.xxx.xxx)(port=3306)(type=master)：java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.sql.SQLException:无法创建PoolableCo
浏览 1提问于2018-03-14得票数 0
回答已采纳
我正在尝试编写一个数据流作业，它需要处理存储中的日志并将它们写入不同的BigQuery表中。要使用哪些输出表取决于日志中的记录。因此，我对日志进行了一些处理，并根据日志中的值生成了一个键。在此之后，我将日志分组在密钥上。我需要将分组在同一个键上的所有日志写入到一个表中。我正在尝试使用带有可调用的beam.io.
浏览 2提问于2019-07-09得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
