Flink windowAll聚合比窗口进程？
Flink windowAll聚合是一种窗口处理方式，与窗口进程不同。
窗口进程是指在流数据处理中，将数据按照时间或者其他条件划分为不同的窗口，然后对每个窗口内的数据进行聚合操作。窗口进程通常是在流数据处理框架内部进行的，由框架自动管理窗口的创建、触发和销毁。
而Flink的windowAll聚合是一种特殊的聚合方式，它将所有的输入数据都放入一个窗口中进行聚合操作。这意味着所有的数据都会被聚合在一起，而不是按照窗口进行分组聚合。这种聚合方式适用于一些特殊的场景，例如需要对整个数据集进行全局聚合的情况。
Flink提供了windowAll聚合的API和相关的函数，可以方便地进行全局聚合操作。在使用Flink进行windowAll聚合时，可以根据具体的需求选择不同的聚合函数，例如求和、求平均、求最大值等。同时，Flink还提供了丰富的窗口函数和触发器，可以对窗口进行灵活的控制和操作。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云Flink产品介绍：https://cloud.tencent.com/product/flink
- 腾讯云流计算Oceanus：https://cloud.tencent.com/product/oceanus
- 腾讯云数据仓库CDW：https://cloud.tencent.com/product/cdw
- 腾讯云数据湖DLake：https://cloud.tencent.com/product/dlake
请注意，以上推荐的腾讯云产品仅供参考，具体选择还需根据实际需求和情况进行评估。
相关·内容
我是Flink的新手。有时，我想在不需要先做keyBy的情况下在DataStream上进行聚合。为什么Flink不支持聚合(sum、min、max等)在DataStream上？
谢谢你艾哈迈德
浏览 1提问于2021-03-16得票数 1
1回答
在阅读了flink的文档和搜索之后，我不能完全理解flink是如何处理窗口中的状态的。假设我有一个每小时滚动一次的窗口，其中包含一个聚合函数，该函数将msg累积到某个java pojo或scala case类中。该窗口的大小是与在一个小时内进入该窗口的事件的数量相关，还是仅仅与pojo/case类相关，因为im将事件累加到该对象中。(例如，如果将10000个msg数成一个整数，大小是否接近10000 *msg大小或int的大小？)另外，如果我使用pojos或case类，flink是否会为我处理状态(如果内存耗尽，会溢出到磁盘/在检查点保存状态等)，还是必须使用flink的状态对象？ 谢谢你的
浏览 49提问于2019-03-20得票数 5
回答已采纳
2回答
闪烁和非键控窗口状态
、、、、
我正在创建一个Flink应用程序，它简单地将窗口传入的Kafka事件转发到另一个Kafka主题，并为每个窗口添加开始和结束标记-例如，对于一个包含1, 2, 3, 4, 5的1小时窗口，我会将start_timestamp, 1, 2, 3, 4, 5, end_timestamp插入到不同的Kafka主题中。可能，稍后还会有其他一些转换，但一般来说，对于传入的N个事件，我将始终至少发出N+2事件。 据我所知，使用带有注入开始和结束标记的ProcessAllWindowFunction的windowAll()应该可以做到这一点。我的问题是关于国家管理的。我将使用RocksDb状态后端-即使对
浏览 39提问于2021-02-03得票数 0
回答已采纳
我有一个高度并行化的聚合，我在多个节点上运行了很多密钥。然后，我想对所有类似于下面代码的值进行汇总：
val myStream = sourceStream
.keyBy( 0 )
.window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
.reduce(_ + _)
.addSink(new OtherSink)
val summaryStream = myStream
.map(Row.fromOtherRow(_))
// paralle
浏览 3提问于2017-08-05得票数 3
我想知道，无论如何，Flink的数据流API是否可以用来从传入的记录中删除重复项(可能是在特定的时间窗口内)，就像Dataset API一样，它提供了一个称为"Distinct“的转换。或者，如果数据集被转换为数据流以便在Flink中进行内部处理，则无论如何都可以将该数据集转换为数据流。
在这方面请帮帮我。提前感谢！干杯!
浏览 1提问于2020-03-05得票数 0
我对弗林克很陌生。我阅读了文档并播放了一些示例代码，但我很难开始处理我的需求。
我们希望使用Flink来维护窗口聚合，作为事务监视应用程序的一部分。它们将使用滑动窗口定义。例如：“过去5天的现金交易总额”。下面是我需要Flink应用程序做的工作: 1.通过读取历史聚合和构建windows 2来准备事务处理。对于每个新事务: a.用新的事务数据b更新窗口聚合。找到与传入时间戳匹配的窗口，并将聚合值添加到事务c。通过RabbitMQ或Kafka接收器将增强的事务(原始字段+聚合)发送到下游处理器。
对于进入的每个事务，我想要一个(而且只有一个)输出，其中包含原始事务字段和聚合。
我了解了如何执行代
浏览 5提问于2017-06-23得票数 0
2回答
我正在尝试编写一个Flink应用程序，它从Kafka中读取事件，从MySQL中丰富这些事件，并将这些数据写入HBase。我正在RichFlatMapFunction中进行MySQL丰富，现在我正在尝试找出如何最好地写入HBase。我想批量写入HBase，所以我目前正在考虑使用WindowAll，然后使用identity apply (只是为了返回DataStream)，然后编写一个HBaseSink，它获取一个记录列表并对HBase执行批量Puts。
这是做事情的正确方式吗？使用WindowAll和apply来做基于时间的缓冲感觉很奇怪。
浏览 0提问于2019-09-27得票数 0
我是弗林克的新手。我有一个Flink流媒体程序，在10秒的会话窗口中计算出卡夫卡的一些内容。
以下是我的问题：
会话窗口默认触发器为FIRE。
Flink流会在内存中保留每个窗口的全部kafka消息吗?还是只保留聚合的结果。
如果我使用FIRE_AND_PURGE，什么会被清除？
浏览 4提问于2019-02-19得票数 2
回答已采纳
1回答
我对flink驱逐者与AggregateFunction和ReduceFunction等热切评估的窗口函数之间的关系感到困惑，
来自flink文档：()
驱逐者能够在触发器触发后和应用窗口函数之前和/或之后从窗口中移除元素。
听起来驱逐者并不急切，那么使用带有AggregateFunction的驱逐者的行为是什么呢？在内存中的窗口中是否有分配所有元素的时间？
谢谢
浏览 0提问于2019-05-06得票数 0
回答已采纳
我在github存储库中搜索了RichAggregateFunction，如下所示。
.aggregate() does not support [[RichAggregateFunction]], since the reduce function is used internally in a [[org.apache.flink.api.common.state.AggregatingState]].
是否意味着Flink不能合并组窗口中元素的状态？
浏览 18提问于2018-02-24得票数 1
1回答
我们正在将火花作业迁移到flink。我们已经在火花中使用了预洗牌聚合。是否有办法在火花中执行类似的操作。我们正在消耗来自apache kafka的数据。我们正在使用键控滚动窗口来聚合数据。我们希望在执行洗牌之前将数据聚合到flink中。
浏览 11提问于2021-08-17得票数 3
回答已采纳
我在Flink中有一个流，它从一个源发送立方体，对立方体进行转换(将立方体中的每个元素加1)，最后将其发送到下游以打印每秒的吞吐量。
流在4个线程上并行化。
如果我理解正确的话，windowAll操作符是一个非并行转换，因此应该将并行化缩小到1，并将其与TumblingProcessingTimeWindows.of(Time.seconds(1))一起使用，求出最近一秒内所有并行化的子任务的吞吐量总和，并打印出来。我不确定我是否得到了正确的输出，因为每秒的吞吐量是这样打印的：
1> 25
2> 226
3> 354
4> 372
1> 382
2> 403
浏览 0提问于2018-06-13得票数 1
除了WindowAll操作之外，还有其他方法可以有效地对DataStream进行排序吗？
让我们以Flink示例中定义的页面视图的用例为例。此示例尝试聚合每15分钟窗口的页面查看计数。
如果我想要改变这一点以获得15个窗口的前3个页面浏览量，那么有效的方法是什么？
一种选择是在聚合函数之后使用windowAll函数，并在内存中进行排序。问题是- windowAll会将并行度降低到1，并且它要求将所有数据放在同一个任务槽中以执行排序操作。理想情况下，当我们只有几个不同的键(即不同的页面URL)时，这是很好的。在我的用例中，在15分钟的窗口内将有数百万或数十亿个键，因此，所有这些百万或数十亿聚合的
浏览 0提问于2021-02-15得票数 0
我们已经开发了一个Flink应用程序，它接收具有字段的消息，例如Flink应用程序中的id, value1, value2, ...，它们由id键控，并分配不同大小的滑动窗口和10秒的slide。我们使用AggregateFunction为每个id计算一些统计信息，结果数据流被汇聚到Redis。代码如下所示： DataStream<Tuple2<String, String>> statistics = messages.keyBy(0)
.timeWindow(Time.seconds(300), Time.seconds(10))
.aggrega
浏览 12提问于2021-01-12得票数 1
回答已采纳
我想更新我的flink应用程序1.11.0 -> 1.13.2我的代码库由于StreamExecutionEnvironment.setStreamTimeCharacteristic，timeWindow()的折旧而停止编译
一些代码
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(20,10000L))
env.getConfig.setGlobalJobPara
浏览 29提问于2021-08-16得票数 0
回答已采纳
1回答
我要把两个关键的溪流连接到一个窗口上。但是，我需要首先减少这两个流，然后在聚合上应用连接，所有这些都在同一个窗口中。
我花了一段时间研究Flink API，发现这样的应用函数存在于WindowedStream中。
公共SingleOutputStreamOperator应用(reduceFunction reduceFunction，WindowFunction函数)
但我无法在JoinedStream中找到类似的函数。问:这会对Flink有帮助吗？有什么办法我现在可以用吗。
浏览 6提问于2016-06-02得票数 1
回答已采纳
当事件时间水印时，Flink如何处理延迟的元素，我有些困惑。
我的理解是，当Flink读取数据流时，水印时间是在看到任何事件时间大于当前水印的数据时进行的。然后，任何覆盖时间严格少于水印的窗口都会触发驱逐(假设不是延迟允许的话)。
然而，以这个最小的例子为例：
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment}
imp
浏览 3提问于2018-05-01得票数 3
回答已采纳
1回答
我想对我的程序(非键控数据)使用滚动窗口功能，因为它正在处理流数据，但只有300条消息/秒。我想把它带到至少5K/秒。为了达到这个目的，我想在2秒内使用翻滚窗口来提高它的性能。但我不知道如何在我的情况下使用这个。
注意:我正在使用Geomesa HBase平台保存消息。另外，我没有在这里粘贴我的整个应用程序代码，因为我只需要窗口函数，这里的代码对您的理解是足够的。
这是我的flink代码
public class Tranport {
public static void main(String[] args) throws Exception {
// fetch
浏览 2提问于2021-08-19得票数 0
我有卡夫卡的一系列数据：
stream
.keyBy("key")
.window(<tumbling window of 5 mins>)
.aggregate(<aggFunc>, <function adding window key and start wd time>)
...
.addSink(sink)
一个窗口中的键数是动态的(窗口1有3个键，窗口2可以有4个键)。
窗口1输出:5分钟- 00:00到00:05，例如。编号只供以后参考，每条记录都是POJO，这里我只保留重要的字段。
1.
浏览 0提问于2018-12-04得票数 2
回答已采纳
我正试图利用并行性来加快前10名窗口操作的速度。我的应用程序由具有时间戳和键以及(即Tuple2<Long,String>)的事件组成，我的目标是为30分钟的滚动窗口(使用事件时间)生成前10位最频繁的键。为此，我的查询包括一个入口、一个窗口和一个聚合阶段。换句话说，我的代码需要执行如下操作：
DataStream<Tuple3<Long, String, Integer>> s = env
.readTextFile("data.csv")
.map(new MapFunction<String, Tuple3<
浏览 1提问于2017-05-26得票数 1
回答已采纳
我在这里读到了有关flink`s的窗口赋值器的文章：https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/windows.html#window-assigners，但我找不到任何解决我的问题的方法。 作为我的项目的一部分，我需要一个窗口，即计时器将启动给关键的第一个元素，并将被关闭，并设置为X分钟后的处理准备。例如： 第一个keyA出现在(hh:mm:ss) 00:00:02，我希望所有的keyA都是窗口，直到00:01:02，然后只有当keyA作为输入时，1分钟的计时器才会再次开始。 有没有
浏览 27提问于2019-12-24得票数 0
1回答
我的要求是保持30天的数据流到给定的任何一天进行处理。因此，当FLINK应用程序启动的第一天，它将从数据库中提取30天的数据，并将合并到当前流数据。我的挑战是-管理30天数据窗口。如果我创建30天的滑动窗口，滑动时间为1天。就像这样
WatermarkStrategy<EventResponse> wmStrategy = WatermarkStrategy.<EventResponse>forBoundedOutOfOrderness(Duration.ofMillis(1))
.withTimestampAssigner((event
浏览 41提问于2020-07-30得票数 0
1回答
我正在处理来自IOT设备的一系列事件。
这些事件具有由网络设置的第一级时间戳。他们还把在不同时间点采取的几项措施结合在一起。例如：
网络时间9:08测量M1在8:52测量M2在9:07
度量是每小时聚合的，在这种情况下，M1应该在8:00-9:00窗口中，M2应该在9:00-10:00窗口中。
我想知道如何设计我的flink应用程序，管理那些时间戳，以及相关的水印。根据我迄今为止的理解：
I可能应该将与网络时间相关的所有处理(9:08)放在一个单独的Flink应用程序中。有一个Flink应用程序来处理解压后的措施(Flink映射)。然后用assignTimestampsAndWatermark
浏览 2提问于2021-04-19得票数 0
回答已采纳
我们在1分钟内聚合一些数据，然后将这些数据刷新到一个文件中。数据本身就像一个地图，其中key是一个对象，value也是一个对象。 由于我们需要一起刷新数据，因此我们没有执行任何keyBy，因此使用了windowAll。 我们面临的问题是，如果我们在ProcessAllWindowFunction中使用窗口函数，然后在进程调用中聚合，那么当我们使用聚合和窗口函数时，我们会获得更好的吞吐量。当我们使用aggregate时，我们也会在状态检查点中看到超时。 我试着浏览了代码库，我能想到的唯一假设是，对进程将使用的ListState与聚合将使用的AggregateState进行检查点可能更容易。 假
浏览 20提问于2021-04-19得票数 0
回答已采纳
1回答
我的Flink工作必须在每一次工作轮班后计算一定的聚合。轮班是可配置的，看起来类似于：
1st shift: 00:00am - 06:00am
2nd shift: 06:00am - 12:00pm
3rd shift: 12:00pm - 18:00pm
为了操作的目的，每天轮班是一样的，一周/一年中的几天是没有区别的。移位的配置可以随时间而变化，而且可以是非单调的，因此这就从表中忽略了一个简单的EventTime窗口，比如：TumblingEventTimeWindows.of(Time.of(6, HOURS))，因为其中一些班次可能会缩小或跨越加班，或者在中间插入几个小时.
我想出
浏览 2提问于2020-11-06得票数 0
我写了一个Flink程序，从一个简单的kafka流计算每个键控窗口的事件数。当源停止时，我希望将0作为每个窗口的计算结果，但没有发送任何结果。我认为这是因为Flink的懒惰操作行为。
有什么建议吗？
浏览 1提问于2015-12-17得票数 1
1回答
我有一个物体坐标流(time，x，y)，我想把它转换成一个距离流，然后再转换成一个速度流。为此，我每次需要处理两个连续的消息。
你能给我一个关于如何创建点之间的欧几里得距离流的想法吗？
浏览 10提问于2018-02-19得票数 0
回答已采纳
在flink中定义时间窗口后，如下所示：
val lines = socket.timeWindowAll(Time.seconds(5))
如何计算5秒的特定窗口中的记录数？
浏览 0提问于2017-08-10得票数 2
因此，这两种方法的代码是相同的，大致如下：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// prepare the topology...
env.execute();
场景是Flink在一台机器上本地运行。
独立罐
pom.xml (相关bits)：
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core<
浏览 1提问于2019-11-18得票数 1
回答已采纳
我对流/事件处理和Flink有些陌生，我遇到了以下问题。
我正在处理来自Kafka的发票事件，这些事件有一个事件“时间戳”和一个"scheduledAt“日期(时间戳)，并希望在发票”scheduledAt“上执行每日聚合。在传统的数据库中，我会这样做：
SELECT AVG(total)
FROM "Invoice"
WHERE date_trunc('day', "scheduledAt") = date_trunc('day', CURRENT_TIMESTAMP)
然而，在流上下文中考虑这一点时，我尝试使用'
浏览 9提问于2018-02-08得票数 0
我有一个ProcessAllWindowFunction实现(请参阅下面代码中的AttributeBackLogEvents() )，它有相当多的I/O，可能需要超过30秒。windowAll()正在使用30秒的TumblingProcessingTimeWindows对数据进行窗口化。
attributedStream
.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(30)))
.process(new AttributeBackLogEvents())
浏览 2提问于2021-10-01得票数 0
我有一个用例，我想用预先确定的开始时间和结束时间来计算窗口。例如，我想计算4月1日下午1点到4月7日下午1点之间的所有事件 我认为我可以计算每天的窗口，并保持结果的状态，但这似乎是最优的。 Flink是否支持预定开始和结束时间的时间窗口？
浏览 10提问于2021-04-12得票数 0
回答已采纳
我的Flink任务的检查点越来越大。在深入到各个任务之后，键控窗口函数似乎要对大部分大小负责。我怎么才能减少这个呢？
浏览 1提问于2021-03-04得票数 1
我们计划使用Apache Flink对多种类型的对象执行实时聚合。我们需要支持几种类型的聚合，如sum，max，min，average等-到目前为止没有什么特别的
我们的需求是将数据输出到kafka，其中一条消息包含多个对象属性的多个聚合值。
例如，消息应包括属性A的总和、最大值和平均值，以及最近10分钟内属性B的总和和最小值
我的问题是，用Flink实现这种要求的最佳方式是什么？
我们考虑使用一个自定义窗口函数，该函数将在窗口末尾的所有对象上运行，并自行计算所有必需的值，并输出一个包含所有这些聚合值的新对象。我们关心的是这个解决方案对内存消耗的影响，必须将所有窗口数据保存在内存中，等待窗口触
浏览 3提问于2017-06-05得票数 4
我正在专门评估Flink的流式窗口支持，以支持可能的警报生成。我关心的是内存的使用，所以如果有人能在这方面提供帮助，我将不胜感激。
例如，此应用程序将在给定的翻滚窗口内潜在地消耗大量数据。在评估时，例如，如果有一百万个文档符合条件，它们是否都会加载到内存中？
一般的流程是：
producer -> kafka -> flinkkafkaconsumer -> table.window(Tumble.over("5.minutes").select("...").where("...").writeToSink(someKafka
浏览 0提问于2017-11-04得票数 2
我需要在flink中有聚合窗口。我不能使用聚合函数。因为getResult计算需要我的状态访问。因此，我尝试在process中使用aggregate：
.aggregate(
new AggregateFunction<Entry, Double, Double>() {
...........
};
, new ProcessWindowFunction<Double,Result,Entry,TimeWindow>() {
浏览 20提问于2020-06-03得票数 0
回答已采纳
例如，我有一大串字，想数每一个字。问题是这些词是不正确的。这意味着有些词的频率会很高，但大多数其他单词的频率很低。在风暴中，我们可以用以下方法来解决这个问题。首先在流上做洗牌分组，在每个节点中在窗口时间内计数本地单词，最后更新计数到累积结果。从我的另一个中，我知道Flink只支持键控流上的窗口，否则窗口操作将不是并行的。
我的问题是，是否有一个好的方法来解决这种扭曲的数据问题在Flink？
浏览 2提问于2016-01-08得票数 6
回答已采纳
目前，我正在使用Flink进行流处理引擎的研究。在我的学习中，我使用由以下形式的元组组成的历史流：
event_time, attribute_1, ..., attribute_X
在处理过程中，event_time作为TimeCharacteristic.EventTime使用。此外，我通过以下方法将数据集推入处理拓扑：(i)创建内存中的结构，或(ii)读取CSV文件本身。
不幸的是，我已经注意到，即使有足够多的元组到达一个完整窗口的窗口操作符中，也不会将该窗口推到下游进行处理。结果，性能显著下降，而且很多时候我有一个OutOfMemoryError异常(具有较大的历史流)。
为了演示一个
浏览 0提问于2018-02-06得票数 0
回答已采纳
1回答
我想发出一个时间窗口的最后一条记录。这可以在常规的Flink中使用maxBy轻松完成，但我无法通过SQL API使其工作。我想要的是：
SELECT LAST(attribute) FROM [table]
GROUP BY key, TUMBLE(ts, INTERVAL '1' DAY)
它的行为类似于
ds.keyBy(key)
.window(TumblingEventTimeWindows.of(Time.days(1)))
.maxBy(x -> x.getTs())
有没有办法在SQL API中实现这一点？
浏览 21提问于2020-05-26得票数 0
我正在寻找标记在我的流中的数据到达flink的时间，以便我可以执行一些计算。我认识到当使用事件时间模型时，我可以直接控制它，但我希望有一些简单的方法来发现flink在对流进行窗口决策时使用的时间戳flink。
浏览 3提问于2017-09-30得票数 0
回答已采纳
我有一个DataStream，需要在它上面计算一个窗口聚合。当我执行常规窗口聚合时，网络IO非常高。因此，我希望执行本地预聚合，以减少网络IO。
我想知道是否有可能在本地对任务管理器进行预聚合(即，在混洗记录之前)，然后执行完整的聚合。使用Flink的DataStream应用程序接口可以做到这一点吗？
我的代码是：
DataStream<String> dataIn = ....
dataIn
.map().filter().assignTimestampsAndWatermarks()
.keyBy().window().fold()
浏览 25提问于2017-12-15得票数 2
我正在使用Flink 1.1.4，并在flink-conf.yaml中添加了用于内存调试的配置参数，如中所述
taskmanager.debug.memory.startLogThread: true
taskmanager.debug.memory.logIntervalMs: 1000
重新启动Flink之后，我将看到添加到Job界面的新参数，但是我无法看到任何新的日志。
知道我可能错过了什么吗？
浏览 4提问于2017-01-24得票数 5
我有一个像这样的流：<_time(timestamp), uri(string), userId(int)>。_time属性是rowtime，我将它注册为一个表：
tableEnv.registerDataStream("userVisitPage", stream, "_time.rowtime, uri,userId");
然后我查询表：
final String sql =
"SELECT tumble_start(_time, interval '10' second) as timestart,
浏览 1提问于2018-03-02得票数 1
1回答
源：动态数据流
Sink：弹性搜索
都可以使用AWS服务。
另外，在AWS动态数据分析应用程序上运行我的Flink工作
我面临一个问题，窗口功能的flink。我的工作看起来像这样
DataStream<TrackingData> input = ...; // input from kinesis stream
input.keyBy(e -> e.getArea())
.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
.reduce(new
浏览 7提问于2021-12-13得票数 0
回答已采纳
我有一些困难，了解窗口是如何在内部实现Flink，并没有找到任何文章，以解释这一点的深入。在我看来，有两种方法可以做到。考虑下面一个简单的窗口字计数代码
env.socketTextStream("localhost", 9999)
.flatMap(new Splitter())
.groupBy(0)
.window(Time.of(500, TimeUnit.SECONDS)).sum(1)
方法1:将所有事件存储500秒，在窗口的末尾，通过对存储的事件应用和操作来处理所有事件。
方法2:我们使用一个计数器来存储每个窗口的滚动金额。当窗口中的每个事
浏览 6提问于2022-01-17得票数 0
回答已采纳
1回答
我看到了一篇文章，在帖子里，公司里的人叫Klaviyo，在不同的时间框架内做计数，每小时，每天，甚至每月。
我有几个问题，如果我正确理解，他们使用的是时间窗口，但使用时间窗口这么长的时间，就像一天一样正常吗？！
这对我来说是没有意义的，如果你每天或每月进行计数，为什么不使用批处理呢？在这种情况下使用流的基本好处是什么？
一个不同的例子，如果我从一开始就需要计算卡夫卡事件，实时的，，什么是真实世界的解决方案？每次事件到达时，使用flink流更新redis中的“计数器”？如果卡夫卡非常繁忙，就像每秒几百万条信息，难道没有太多的IO和网络吗？
浏览 0提问于2019-03-28得票数 0
回答已采纳
1回答
如何在Apache Beam中创建基于元组的滑动窗口？这在Flink中很容易做到：
DataStream.countWindowAll(long size, long slide)
但从Beam (或DataFlow)的文档中还不清楚如何做到这一点。它是窗口和触发器的某种组合吗？它是否有效？
浏览 2提问于2017-05-01得票数 0
全新的beam/flink，希望在这个问题上提供帮助。我有一个管道，从卡夫卡，阿夫罗的信息，做一些对象转换，并再次写到卡夫卡。ו没有定义任何窗口，因为目前我们希望在没有聚合的情况下单独处理每个事件。我想知道这是否正确。根据我在文档中的理解，我们似乎不能使用默认行为并定义某种类型的窗口和相关触发器。
我的理解正确吗？谢谢S
浏览 0提问于2022-04-25得票数 0
1回答
当flink会话窗口在一段时间内没有接收元素时，我使用的是flink会话窗口，也就是说，当不活动的间隙发生时，它应该会发出一个事件。
我在flink作业中将间隙配置为10秒。我发送event1并在5秒后发送event2。这两个事件应该属于第一个窗口。输出应该是这两个事件的集合。但我只得到了第一件事。
下面是我尝试过的代码：
fun setupJob(env: StreamExecutionEnvironment) {
val testStream = env.sampleStream()
.keyBy { it.f0 }
浏览 5提问于2022-02-23得票数 0
回答已采纳
我试图转换数据流，而不使用Flink提供的任何窗口。我的代码如下所示：
val stream1 = executionEnvironment.getStream
val stream2 = stream1.flatMap(someFunction)
stream2.addSink(s3_Sink)
executionEnvironment.execute()
但是，在提交和运行我的工作时，我无法在S3上获得任何输出。web显示接收到的0字节、接收到的记录、发送的0字节、发送的记录。
另一个正在运行的Flink作业已经使用了相同的数据源，因此数据源很好。任何地方都没有错误，但仍然没有输出。这个
浏览 0提问于2019-04-03得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
