我使用的是apache beam 2.17.0的Python 3.7SDK的数据流。代码在本地运行，但我从pubsub收集数据。在调试和深入源代码之后，我发现任务没有正确完成，这就是为什么它被执行了两次。) |'Filter metatags'
浏览 7提问于2020-05-13得票数 2
我使用的子类来实现CombineFn，而不是使用SerializableFunction的实现import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.transforms.Combine.Combin
浏览 0提问于2019-06-04得票数 1
回答已采纳
我正在创建一个Apache光束流处理管道，以便在GCP数据流中运行。我有许多扩展DoFn和CombineFn的转换。在DoFn中，使用数据流作业详细信息中的日志窗口可以很好地显示日志。但是，不会显示来自CombineFn转换的日志。 我尝试了不同的日志级别，但使用DirectRunner时，它们也显示得很好。 下面是一些示例代码。import java.io.Serializable;
import org.
浏览 15提问于2019-10-07得票数 0
我正在尝试将一个包含多个元素的PCollection转换为一个只有一个元素的PCollection。基本上，我想说的是：至：这样我就可以在一个DoFn中处理整个DoFn了。我尝试过CombineGlobally(lamdba x: x)，但一次只有一部分元素被组合到数组中，给出了以下结果：或者其他类似的东西这是我正在尝试运行的脚本的</
浏览 5提问于2022-08-01得票数 1
回答已采纳
2回答
：如何在中实现MapState？
、、、、
我正试图在apache中创建一个有状态的ParDo，它存储一个值块，并使用后续窗口中的数据进行更新。在java中等效为。我尝试使用自定义CombineFn来实现它。class DictCombineFn(beam.CombineFn): return {}
class EnrichDoFn(beam.DoFn)
浏览 1提问于2019-05-17得票数 2
我有一个(键，值)对的PCollection，其中值本身是一个列表 {'apple': ['1','2'], 'apple': ['9','10']} 我希望获得另一个具有相同键的集合，但是列表是原始集合上的GroupByKey，但是结果值是扁平化的。我猜人们总是可以编写一个ParDo函数，在每
浏览 20提问于2021-10-18得票数 2
回答已采纳
index = i; }和MergeCellsFn equality constraints: org.apache.beam.sdk.values.PCollection<org.apache.
浏览 0提问于2020-08-22得票数 0
我目前正在尝试删除Google云存储桶中的所有0字节文件。我希望能够通过apache和运行在google云项目上的数据流运行器来完成这个任务。我现在拥有的是(我用<***>隐藏了一些细节)：import apache_beam.io.gcp.gcsfilesystem as gcs
fromapache_b
浏览 2提问于2021-02-07得票数 1
回答已采纳
2回答
我们目前正在尝试一条简单的管道，以便：
at org.apache.beam.sdk.transforms.Combine)
at org
浏览 2提问于2017-05-16得票数 9
回答已采纳
1回答
每当新窗口开始时，直到第一个窗口完成为止，都会触发。如何只发出一次转换后的数据或避免重复？
我使用过废弃的烧制窗格，这并没有什么区别。线程"main“org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.IllegalArgumentException中的异常(DirectRunner.java:64) at org.apache.beam.sd
浏览 0提问于2019-07-17得票数 0
使用Google的python，我已经重写了CombineFn来完成多字段聚合，而不是一次一个字段。根据文档，add_input的接口是分组后的中间聚合和输入值。此功能在本地运行程序场景中运行得很好。
当我移动到google引擎时，我看到了一组具有上述条件的日志，它完成了id1和id2的聚合。然后，我突然看到了对add_input的另一个调用，在前面的计算中，intermediate_value被用
浏览 1提问于2018-07-11得票数 1
回答已采纳
根据
"--environment_type=PROCESS", ] &
浏览 18提问于2022-06-02得票数 1
回答已采纳
我的团队在许多用于监视和分析的数据流管道中使用了大量的聚合器(自定义计数器)。我们编写DoFn类主要是为了做到这一点，但有时我们通过编写自己的实现SerializableFunction<Iterable<T>, S>的组合类来使用Combine.perKey() (在我们的例子中看来聚合器只能在DoFn中使用，我想知道是否有办法解决这个问题，或者这是将来可能添加的特性。通常，我们使用一组自定义计数器来统计不同类型的</
浏览 5提问于2016-11-15得票数 0
回答已采纳
2回答
"datapointN" : "Some Integer",在我的示例中，将有数千个具有全名"{location}_{name}“的传感器。对于每个传感器，我希望将数据窗口设置为60秒，并计算该数据的平均值。为了插入Bigtable和Bigquery，我将把这个最后的形式作为1元素存在。然后，我运行一个ParDo来处理这个问题，将json分解成(键，值)元组，为CombinedValues做好准备，所有的数据再次
浏览 7提问于2021-01-26得票数 0
回答已采纳
这是我目前所得到的(Apache Beam Python SDK 2.10.0)： import apache_beam as beam
return "\n".join(e)
o = beam.optio
浏览 6提问于2019-02-18得票数 7
回答已采纳
1回答
Apache依赖关系错误
、、、、
我正在尝试使用来自Java的Apache作为数据管道。我编写了一个简单的类，它来自Google并深入到，但我无法让它为我的生命构建。我使用Maven构建并添加了我能找到的每一个束包，但是我仍然会得到“类文件未找到”错误。不幸的是，谷歌一直没有提供任何帮助。
我所拥有的POM文件是基于Apache为Wordcount提供的示例POM，但是添加了额外的依赖项。下面是我输入的依赖项。我可以提供完整的文件，
浏览 0提问于2018-02-12得票数 2
回答已采纳
3回答
是否有可能在DataFlow管线完成后获得回调？流水线完成后，我必须对系统进行一些配置更改，以使用流水线生成的新输出以及其他一些清理工作。因此，当测试在本地系统中运行时，它会暂停开发人员的命令提示符，或者用户必须等待管道完成。
那么有没有更好的方法呢？就像回调机制一样？
浏览 0提问于2019-08-22得票数 1
我正在尝试使用python中的Apache将BigQuery表记录作为JSON文件写入GCS桶中。我有一张像这样的BigQuery表-- my_project.my_dataset.my_table我希望将表记录/条目写入GCS桶位置中的JSON文件中- "gs://my_core_ bucket3333333333"}},
{"id":"4"
浏览 12提问于2021-07-12得票数 1
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
