我正在尝试学习Java中的Apache光束，但是我被卡住了，没有任何进展！796656IE,Kyle,10,Accounts,1-01-2019560447WH,Olga,20,HR,1-01-2019 我想知道如何计算包含"Accounts“的行中的人数，并实现以下输出： Kyle=2
Kumiko=1
浏览 8提问于2020-11-20得票数 1
回答已采纳
在做了一些处理和按键分组之后，我就有了如下所示的数据集。现在，我需要对数据的每一行进行一些处理，以获得下面的输出。我试过平面图，它真的很慢，因为“值”列表的长度可以任意长。如何将每一行分割成不同的pcollection？如果这不可行，还有其他方法可以加快计算速度吗？输入1 [A, B, B, B]3 [A, B, B, B]5 [A, B, B,B]
输出</em
浏览 1提问于2021-03-12得票数 3
回答已采纳
1回答
我在apache beam中的输出是('key'，{'id':name})如何在apache波束中使用Map进行转换以获得预期的输出
浏览 2提问于2021-11-17得票数 0
Steel,graystone,0,8-Speed Dual-Clutch,2018-02-05 01:49:47 UTC,1646,22550,0,28453
我想要建立一个光束管道，它将从csv文件中读取此数据，抓取vin并计算vin在文件中出现的次数。所以我想按vin分组并计算计数。我希望我的最终输出在平面文件中。我错过了注释，所以我现在添加了它，但我得到了一个不同的错误，我也找不到解决方案。下面是我的代码。import org.apache.<
浏览 0提问于2018-10-22得票数 1
2回答
用编写每个窗口的唯一拼花文件
、、、、
我试图用apache将从kafka消费者到google云存储的消息通过30秒的 windows传输到谷歌云存储。使用beam_nuggets.io阅读卡夫卡的主题。但是，我无法在每个窗口都为GCS编写唯一的拼花文件。您可以看到我的代码如下：from apache_beam.transforms.trigger import AfterAny, AfterCount, AfterP
浏览 3提问于2021-10-18得票数 2
在beam 中，似乎没有提到这种情况下的错误处理。关于错误处理的中介似乎不适用于这种返回PDone的特定类型的PTransform。
的答案是不可理解的，没有例子。在我的示例中，我正在读取一个包含两个重复行的文件，并试图将它们写入一个表中。entry 0 INSERT INTO foo (field) VALUES ('a') was aborted: ERROR: duplicate key value viol
浏览 0提问于2019-05-31得票数 2
回答已采纳
目的：问题：如果我减少了元素大小(元素计数不会改变)，它就能工作！因为实际上，按步骤分组等待所有数据被分组，然后触发所有新的窗口数据。我用这两种方法测试：beam版本2.6.0和scio版本0.6.1
读取包含文件
浏览 3提问于2019-04-12得票数 4
我想使用Apache中的流管道(并在Google DataFlow上运行它)来实现以下场景：
我试图同时使用Java(Scala)和Python来解决这个问题，但是解决方案没有奏效Ja
浏览 5提问于2019-03-28得票数 1
will be parsed as list of HashMap objets for each key/value.我在JSON和publishDate中还有一些属性，这些属性将添加到tableRow和每个消息属性中。所以表如下所示。但是，不确定它将如何根据消息列表输出多个行。Context.output(行)在messages.forEach中，在将表行设置为
List<Message>
浏览 3提问于2017-10-26得票数 3
回答已采纳
我正在使用Apache，并且很难以正确的格式按键进行分组/组合，以获得一个PCollection。在字符串的PCollection中，每个键有多个记录，如下所示：Bob, dogCarla, bunny我想使用GroupByKey我可以使用ParDo来获得字符串键和字符串值，如下所示，但我不知道如何转换为KV<string、stri
浏览 8提问于2022-03-23得票数 1
回答已采纳
1回答
从BigTable到模式问题？
、、、、
我正在尝试使用Dataflow模板1将一个Avro文件(使用Spark3.0生成)摄取到BigTable中，并获得以下错误。知道吗？(MapTaskExecutor.java:77)at org
浏览 6提问于2020-12-22得票数 1
回答已采纳
1回答
$FinalizeTempFileBundles$FinalizeFn.process(WriteFiles.java:850) 流式作业是从apache pulsar源获取数据，并将输出数据以parquet我正在使用Spotify的scio在Scala中编写这项工作。我可以看到源路径和输出路径应该具有相同的URI方案，但是我不知道这是如何发生的，因为我使用了一个alluxio路径作为输出目录。在a
浏览 24提问于2020-09-01得票数 0
3回答
如何使用Apache框架按降序排序？ .apply("ExtractWords-> String.format("
浏览 2提问于2017-12-11得票数 3
回答已采纳
1回答
在流式处理过程中，我尝试将json文件发布到pubsub，并使用cloud Dataflow写入数据存储。_namespace
File "<
浏览 0提问于2018-10-12得票数 1
我想使用Dataflow将数据从发布/订阅移动到GCS。因此，基本上我希望Dataflow在固定的时间内(例如15分钟)积累一些消息，然后在该时间段过去后将这些数据作为文本文件写入GCS。我的最终目标是创建一个自定义的管道，所以“发布/订阅到云存储”模板对我来说是不够的，而且我对Java一无所知，这让我开始在Python中进行调整。res = p.run()
res.wait_until_finish() 我在本地环境中运行此程序时
浏览 6提问于2019-02-18得票数 7
回答已采纳
通过错误获得有关此大小问题的通知时，将提供以下信息： the size of the serialized JSON representation of the pipeline exceeds the现在使用此参数，确实会导致dataflow runner将额外的dataflow_graph.pb文件上载到通常的pipeline.pb文件旁边的登台位置。我证实它确实存在于gcp存储中。但是，gcp数据流中的作业在启动后立即失败，并显示以下错误： Ru
浏览 43提问于2019-04-24得票数 5
回答已采纳
我试图编写一个示例，展示定时器和状态如何在ApacheBeam中工作，并使用beam操场测试这段代码。我读了一个数字列表，将它们以divisibleByTwo和divisibleByThree的形式存储在KV中。然后我用这个KV运行一个statefulDoFn。对于可被2整除的数字，我已经将计时器设置为在x秒之后过期，对于可被3整除的数字，我设置为y秒。但是，测试没有在x和y秒之后打印它们。我在测试用例中</
浏览 4提问于2022-07-20得票数 1
回答已采纳
我有一个现有的带有Java 8、ApacheBam2.27.0、Maven和Dagger 2的Apache项目。 val value = entry.value val ref = object : TypeReference<List<MyObject使用Java 8和<em
浏览 4提问于2021-04-30得票数 0
回答已采纳
为了提供我的问题的最小示例，我尝试实现一个简单的Beam作业，它将String作为一个侧输入，并将它应用到从Cloud中的csv文件中读取的PCollection中。然后将结果输出到云存储中的.txt文件中。到目前为止，我已经尝试过:尝试使用PipelineResult.waitUntilFinish (如in (p.run().waitUntilFinish()) )，更改两个p.run()命令的位置，并尽可能地简化: Unknow
浏览 1提问于2017-04-27得票数 0
回答已采纳
当我试图在我的工作中使用组合器时，我得到了下面的例外
在org.apache.hadoop.mapred.IFile$Writer.append(IFile.java:193)在org.ap
浏览 2提问于2015-12-27得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
