我想为具有早期触发的流水线创建一个单元测试。示例管道如下所示： class CalculateTeamScores(beam.PTransform): return scores\ beam.window.FixedWindows(120),(su
浏览 20提问于2020-07-22得票数 4
1回答
LE: TL;DR；如何在Python中创建无界数据源？有可能吗？" >> beam.ParDo(DummyWindowPrint())我遇到的问题是，GroupByKey之后的操作只有在输入完成后才会启动，但是我希望在WindowInto上使用一个触发器，该触发器将在窗口中的最后一个条目到达后1秒后触发。>
DEBUG:apache_beam.runners.worker.sdk<
浏览 1提问于2020-03-12得票数 2
我有一个有状态的DoFn，它基本上对即将到来的元素进行批处理，当缓冲区达到一定大小时，缓冲区将被清除并将元素插入到BigQuery中。我注意到的是，管道有时会引发异常，该异常并不是停止作业运行。File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py"
浏览 7提问于2020-12-09得票数 1
回答已采纳
我有一个简单的管道，3周前它可以正常工作，但是我已经返回代码来增强它，当我试图运行代码时，它返回了以下错误：代码如下所示，它用于将作业提交到数据流中，而不存在任何问题。import jsonfrom apache_beam</
浏览 6提问于2020-10-23得票数 4
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
我的最终目标是创建一个自定义的管道，所以“发布/订阅到云存储”模板对我来说是不够的，而且我对Java一无所知，这让我开始在Python中进行调整。这是我目前所得到的(Apache Beam Python SDK 2.10.0)： import apache_beam as beam
TOPIC_PATH="projects/<my-projectlocal&#x
浏览 6提问于2019-02-18得票数 7
回答已采纳
我正在尝试运行由本地Flink集群支持的Apache，以便使用卡夫卡主题，如中所述。但是，在将一些消息写入myTopic之后，终端仍然是冻结的，并且我看不到输出文件夹中的任何内容我对Beam/Fl
浏览 1提问于2020-08-30得票数 0
回答已采纳
0回答
Apache光束字段分区
、、、
我想通过一个特定的字段来使用Python来划分Apache Beam中的PCollection。我在Python SDK文档中找到了以下代码，其中对PCollection进行了分区def partition_fn(student, num_partitions):return int(get_percentile(student) * num_partitions / 100)
by_decile = stu
浏览 0提问于2017-11-30得票数 0
目的：问题：如果我减少了元素大小(元素计数不会改变)，它就能工作！因为实际上，按步骤分组等待所有数据被分组，然后触发所有新的窗口数据。逐行扁平(因此它生成大约10,000个)元素
创建数据的键值(使用从1到10的随机整数键
浏览 3提问于2019-04-12得票数 4
1回答
如果我将max_num_records =20放在ReadFromKafka中，我可以看到打印或输出为文本的结果。如果我尝试使用beam.io.WriteToText输出，将会创建一个空的临时文件夹，如: beam-temp-StatOut-d16768eadec511eb8bd897b012f36e97 终端显示： 2.30.0: Pulling from apache/beam_java8_sdk
Digest: sha256:720144b98d9cb2
浏览 40提问于2021-07-07得票数 0
1回答
对于下面的代码，我假设它会产生加窗口的输出，在每个窗口之后有效触发。| "Parse" >> beam.Map(parse_json) beam.window.FixedWindows(5*60),
trigger=beam.trigger.Repeatedly(beam.trigger.AfterProcessingTime(5*60)),这
浏览 2提问于2020-06-11得票数 0
回答已采纳
我已经阅读了Apache Beam Python SDK文档和许多问题，但对一些事情仍然不确定。
是否可以将发布/订阅IO作为非流作业的一部分进行读取？然后，同一作业是否可以使用数据存储IO (当前不支持流)进行写入？我是否可以假设默认的全局窗口和触发器将正确地告诉作业何时停止从发布/订阅读取(当不再写入批量消息时)？或者我需要添加某种类型的触发器/窗口方案，如最大时间或最大元素数量？该触发</e
浏览 2提问于2019-02-21得票数 1
有没有一种方法可以访问存储在Cloud Bigtable中的数据，作为运行Apache Beam管道的输入源？
浏览 6提问于2019-05-09得票数 2
我正在用apache beam实现一个批处理管道，它可以解压缩json文件，对它们进行预处理，然后将它们存储回文件系统中的给定位置。 可以使用ZIP或GZIP算法压缩文件。解压对GZIP文件工作正常，但对ZIP文件解压失败...经过调查，我发现只有GZIP、BZIP2和DEFLATE压缩类型仅在JAVA SDK中受支持，而不存在python实现。
浏览 13提问于2020-12-18得票数 0
我不知道如何在bigquery中高效地完成这项工作，所以我在考虑使用数据流。使用dataflow，我们可以首先加载数据，然后为每条记录创建一个键值对，键是我们想要拆分表的特定列的所有可能值，然后我们可以按键对记录进行分组。所以操作应该是:p| beam.io.Read(beam.io.BigQuerySource()) | beam.map(lambda record：(record‘’splitcol‘，record)) | beam.GroupBy
浏览 9提问于2017-07-13得票数 0
我使用beam sdk 2.4和2.5进行了测试。:245) at org.apache.beam.sdk.Pipeline.replaceat org.apache.beam.sdk.Pipeline.run(Pipel
浏览 0提问于2018-07-14得票数 0
回答已采纳
我想根据json模式文件验证ParDo函数中每个元素的数据。 为此，我需要从本地计算机复制json模式文件，以利用由Python Beam Dataflow SDK创建的Dataflow实例。每个单独的元素表示单独的表的数据(这种不同的元素的变化是26，这意味着可以根据表示表名的元素中的键字段将元素转储到这26个表中的任何一个表中)。我希望这个json模式文件只在Dataflow作业开始时复制一次，然后对已经存储了json模式的元素进行</em
浏览 22提问于2021-02-11得票数 0
基本上，我只重用了iris utils和iris pipeline中的代码，对服务输入做了很小的更改： def _get_serve_tf_examples_fn(model, tf_transform_output/site-packages/apache_beam/runners/worker/sdk_worker.py", line 256, in _execute
response = task()File "/
浏览 49提问于2020-10-26得票数 0
回答已采纳
我试图运行一个简单的beam脚本上的GCP数据流，以便应用一个科学工具包-学习模型的一些数据。数据需要在应用模型之前和之后进行处理。如您所见，我尝试使用新的安装在一个新的虚拟环境中运行。知道怎么修吗？from apache_beam.options.pipeline_optionsWARNING:root:M
浏览 12提问于2020-09-13得票数 0
回答已采纳
2回答
用编写每个窗口的唯一拼花文件
、、、、
我试图用apache将从kafka消费者到google云存储的消息通过30秒的 windows传输到谷歌云存储。使用beam_nuggets.io阅读卡夫卡的主题。您可以看到我的代码如下：from apache_beam.transforms.trigger import AfterAny, AfterCount, AfterProcessingTime, AfterWatermark, Repeatedly
f
浏览 3提问于2021-10-18得票数 2
云服务器
ICP备案
云直播
腾讯会议
实时音视频
腾讯云开发者
