尝试实现一个用于读取Kafka流和聚合会话的Flink作业，由于某种原因，getResult()没有被调用。我看到createAccumulator()和add()被调用了，我希望getResult()也被调用，这样我就可以在目的地接收聚合消息。
浏览 76提问于2019-11-28得票数 1
我需要在一个翻滚的窗口里数数事件。但是，如果窗口中没有事件，我也希望发送0值的事件。 return accumulator + 1L; public AggregateOuterClass.Aggregate getResultBasicCounter<MonitorOuterClass.Monitor>());我读到了一个TiggerFun
浏览 5提问于2020-06-17得票数 1
回答已采纳
我有一个这样的流设置，使用Apache Flink1.4，首先是DataStream，然后是keyBy，然后是window，最后是aggregate
聚合操作的输出是一个AggregateResult对象
浏览 2提问于2017-12-23得票数 2
我需要在flink中有聚合窗口。我不能使用聚合函数。因为getResult计算需要我的状态访问。因此，我尝试在process中使用aggregate： new AggregateFunction<Entry, Double, Double>() {Result> output) throws Exception {
但我仍然只需要aggInput的最新值，而且我看不出有什么理由将这些数据保存在内
浏览 20提问于2020-06-03得票数 0
回答已采纳
最终，我希望在会话窗口打开时，一个接一个地用会话窗口id丰富事件(我不想等到窗口关闭后才发出丰富的事件)。我试着用AggregateFunction来做这件事，但是我不认为merge()能像我预期的那样工作。它似乎是用于合并窗口，而不是窗格(触发器触发)。它似乎从未在我的管道中被调用过。会话窗口id将是落入该窗口的第一个事件的时间戳(由于不保证排序，这可能意味着某些事件可能落入具有较早时间戳的相同会话窗口-我对此没有意见)。assigns sessio
浏览 0提问于2019-06-13得票数 0
我试图在连续数据流上创建一个TumblingWindow，并在窗口中创建聚合。但是由于某种原因，getResult()没有被调用。public class MyAggregator implements AggregateFunction<Event, MyMetrics, MyMetrics> {
} }
@Overr
浏览 4提问于2022-02-10得票数 0
我正在努力理解Flink中的AggregateFunction，它被描述为。它有四种方法，即，
当第一个元素进入新窗口并将进一步使用新创建的实例时，将调用createAccumulator调用add方法来根据定义减少结果，这使用在createAccumulator方法中创建的实例。我对上述方法的理解是否正确？当
浏览 4提问于2021-01-22得票数 2
回答已采纳
2回答
我试图统计不同窗口大小的流中的数据(窗口的大小在steam数据中)，所以我使用自定义的WindowAssigner和AggregateFunction，但状态很大(窗口大小从一小时到30天)有什么问题吗？public boolean isEventTime() { }
public class CountAggregate implements AggregateFunctionAggregateResult a, Aggreg
浏览 0提问于2020-01-06得票数 0
在我的Flink作业中有很多延迟事件，所以将allowedLateness()设置为10分钟(使用TumblingEventTimeWindows并在每个窗口上运行一个复杂的AggregateFunction)
是否有任何只影响后期事件的触发因素？
浏览 0提问于2019-08-28得票数 0
回答已采纳
1回答
现在我想声明一个会话，在X秒的范围内收集与单个会话相同的user +值，并使用最新的时间戳，然后它将只被转发到下一个流一次data.assignTimestampsAndWatermarks....... .window(EventTimeSessionWindows.withGap(Time.minutes(10))) }
浏览 6提问于2020-04-06得票数 0
我试图在我的Flink工作中使用事件时间，并使用BoundedOutOfOrdernessTimestampExtractor提取时间戳并生成水印。但是我有一些输入卡夫卡有稀疏流，它可以长时间没有数据，这使得getResult在AggregateFunction中根本没有调用。我可以看到数据进入add函数。
浏览 28提问于2020-04-03得票数 0
回答已采纳
当使用AggregateFunction在Flink中实现EventTimeSessionWindows时，我无法理解在动态间隙的SessionWindow情况下何时发生合并。setParallelism(parameterTool.getInt("zeus-aggregator-parallelism"))我在聚合器中定义了四个函数：
createAccumulator：--这将创
浏览 3提问于2022-02-27得票数 0
WindowFunction的触发器和ReduceFunction的触发器有什么不同？
浏览 0提问于2018-03-13得票数 0
回答已采纳
我有一个，它计算中一系列事件的平均值。换句话说，我需要在计算之前对数据进行排序，因为序列很重要。我可以用 API来实现这个功能，但是这个函数会在窗口中的每个事件上调用，这在性能上是没有意义的。我也可以用来做这件事，但出于同样的原因，我想避免这样做。谢谢最后我使用了
ProcessWindowFunction可以与ReduceFunction、AggregateFunction或FoldFunct
浏览 0提问于2019-02-05得票数 0
回答已采纳
1回答
输入代码hereNeed，使用flink聚合kafka stream中的数据值，然后输出一个新的主题。
聚合应该在eventtime发生，而不是process time，这意味着数据对象中的时间戳。遵循Flink教程中的示例，使用TumblingEventTimeWindow，但根本不调用聚合getResult方法。如果我更改为TumblingProcessingTimeWIndow，getResult将被调用并将结果推送到接收器。由于它是一个传感器事件</em
浏览 1提问于2019-08-06得票数 0
为了可合并，会话窗口操作符需要一个合并触发器和一个合并窗口函数，例如ReduceFunction、AggregateFunction或ProcessWindowFunction。在堆栈溢出回答中，有人指出
这是否意味着对到达我的流的每个事件都调用了一个createAccumulator()？然后每次当gap小于定义的会话间隙时都调用merge()？在我的实现中，createAccumulator()创建一个对象，在对a
浏览 3提问于2022-08-23得票数 2
回答已采纳
1回答
，看起来像这样： { uid:String .....
} 我需要聚合某个时间范围内的汇总值在第一次刷新之后，我想从内存中清除所有的uid，并立即刷新每个新项。 所以我尝试了这个聚合函数。public class AggFunc implements AggregateFunction<Item, Acc, Tuple2<Integer,List<String>>
浏览 16提问于2020-08-24得票数 0
在获取源流之后，我希望通过复合键和timeEvent滚动窗口聚合事件，然后将结果写入表中。问题是，在应用我的aggregateFunction (只计算clientId的点击次数)之后，我找不到获得每个输出记录的键的方法，因为api返回一个累积结果的实例，而不是相应的键。我没有在累加器中注入输入事件的键，因为我觉得自己不太好。
浏览 0提问于2019-07-08得票数 1
回答已采纳
1回答
从异步方法返回值
、、
我有asmx webservice，我在PCL中使用它，所以我们都知道现在PCL不再支持asmx，它支持wcf webservices。当我调试它时，我发现当getResult发生时，它在到达return语句时进入void getResult。现在有一个只有在事件发生时才被调用的webservice (在本例中)，事件getResult的调用(在本例中)是基于网络的，服务器响应将有一些我们无法控制的
浏览 0提问于2014-10-28得票数 0
我正在尝试在WindowedStream上实现一个reduce，如下所示： .timeWindow(Time.ofnegative += t1.negative;
}我遇到的问题是，当我调用stream.print()时，我得到了许多值(看起来像是每个TwitterSentiments对象一个值，而不是一个聚合对象
浏览 18提问于2018-12-05得票数 2
云服务器
ICP备案
对象存储
云点播
智聆口语评测
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号