Flink :处理数据早于应用程序水印的键控流
Flink是一个开源的分布式流处理框架，它具有处理数据早于应用程序水印的键控流的能力。下面是对该问答内容的完善和全面的答案：
Flink是一种基于事件时间的流处理框架，它允许开发人员以实时方式对数据进行处理和分析。在Flink中，流数据被视为一个不断增长的无界数据流，它可以在事件时间、处理时间和摄取时间上进行处理。
处理数据早于应用程序水印的键控流是指在数据流中，数据的事件时间早于应用程序的水印时间。水印是用于触发时间窗口操作的机制，它告知Flink何时认为数据流中的事件已经到达一定的时间点。通过键控流，我们可以根据数据的键（key）进行数据流的分区和分组，从而实现更精确的流处理和分析。
Flink作为一个强大的流处理框架，具有以下优势：
- 低延迟和高吞吐量：Flink采用了基于事件时间的处理模型，可以处理实时数据，并具有低延迟和高吞吐量的优势。
- Exactly-Once语义保证：Flink提供了精确一次性语义保证，确保每条消息都被处理且仅被处理一次，保证数据的准确性和一致性。
- 支持容错和故障恢复：Flink具有内置的容错机制，当出现故障时能够保证数据处理的可靠性，并能够自动恢复。
- 灵活的流处理：Flink提供了丰富的API和库，使开发人员可以使用各种编程语言和工具来开发和调试流处理应用程序。
- 大规模流处理：Flink可以在大规模分布式集群上运行，具有良好的可扩展性，可以处理大规模的数据量。
- 生态系统和集成：Flink与其他大数据生态系统（如Apache Kafka、Apache Hadoop、Apache Hive等）无缝集成，可以与各种数据存储和处理系统进行交互。
针对处理数据早于应用程序水印的键控流，腾讯云提供了相应的产品和服务：
腾讯云的实时计算引擎CDP（Cloud Data Processing）可以作为处理数据早于应用程序水印的键控流的解决方案。CDP提供了低延迟、高可用和可扩展的流处理引擎，具有精确一次性语义保证和容错机制。通过CDP，用户可以轻松构建和管理基于Flink的流处理应用程序，并实现对数据早于应用程序水印的键控流的处理。
更多关于腾讯云CDP的信息，请参考腾讯云的官方文档： https://cloud.tencent.com/product/cdp
相关·内容
我读了源代码，将每一个结果都减少到下游。
我要按键减少一个没有窗口的流，
stream.keyBy(key)
.reduce((a, b) -> {
//reduce
return a+b;
});
当水印到达时，如果减少窗口，flink将向下游转发元素，因此flink如何确定在没有窗口的情况下减少完成。
浏览 3提问于2019-09-18得票数 3
1回答
我有一个定制的flink Source，还有一个SerializableTimestampAssigner，它将事件时间戳分配给源发出的记录。由于底层数据存储的性质，源可能会发出无序的记录，但是在批处理模式下，我希望在任何操作符处理这些记录之前，Flink会根据事件时间戳对这些记录进行排序。
摘自Flink 的执行模式：
在批处理模式中，在预先知道输入数据集的情况下，不需要这样的启发式，因为至少可以按时间戳对元素进行排序，以便按时间顺序处理元素。
然而，情况似乎并非如此。如果我用时间戳分配器在Source (StreamExecutionEnvironment.fromSource)中
浏览 1提问于2022-04-22得票数 1
回答已采纳
我正在构建一个Flink流应用程序，并且更倾向于使用事件时间，因为它将确保所有已设置的计时器在历史数据发生故障或重播时决定性地触发。事件时间的问题是，只有当事件发生时，时间才会向前移动。我们的数据源(物理传感器)有时产生的数据非常少，因此有时单个数据点可能会打开一个5分钟的聚合窗口，但下一个数据点是20分钟后，因此窗口关闭并很晚发布输出记录。
我们建议的解决方案是使用AWS lambda函数，该函数计划每隔X分钟运行一次，该函数将虚拟事件输出到Flink读取的Kinesis流中，从而强制生成一个水印以向前推进时间。
我担心的是，只有在水印是真正全局的情况下，这才有效，这意味着单个心跳消息可以导
浏览 2提问于2019-12-06得票数 0
回答已采纳
我在使用Flink连接两个流的上下文中查看，并希望了解这两个流之间的区别以及影响Flink处理它们的方式。
作为一个相关的问题，我还想了解CoProcessFunction与KeyedCoProcessFunction有何不同。
浏览 2提问于2021-02-17得票数 1
回答已采纳
这是一个非常广泛的问题，我是Flink的新手，正在研究将其作为当前分析引擎的替代品的可能性。 场景是，从各种设备采集的数据，以JSON编码字符串的形式接收，格式为{“location.attribute”：value，“TimeStamp”：value} 例如，接收位置的单一可追溯性代码，之后在实时流中接收各种过程参数。分析将在过程参数上运行，但是输出需要包括与可追溯性代码的关系。例如{“location.alarm”：value，“location.traceability”：value，“TimeStamp”：value} Flink使用什么方法来缓存值，在这种情况下是当前的可追溯性代码
浏览 17提问于2020-04-11得票数 0
回答已采纳
我有一个带有事件时间的数据流，是否可以在Apache Flink中发出没有数据的窗口？例如，如果翻滚窗口的大小为15 minutes，例如，如果在[(0, 15), (15, 20), ...]的时间段内没有事件，我可以清空窗口，这将是理想的。
浏览 16提问于2019-02-08得票数 0
回答已采纳
我正在使用Flink 1.12并且我有一个键控流，在我的代码中看起来A和B共享相同的水印？因此，因为A的到来已经将水印升级为2020-08-30 10:50:11，所以B被确定为延迟 输出为A(2020-08-30 10:50:08, 2020-08-30 10:50:16):2020-08-30 10:50:15，没有B的输出 我想问，是否有可能使不同的密钥具有独立的水印？A的水印和B‘水印独立变化 应用程序代码为： import java.text.SimpleDateFormat
import java.util.Date
import java.util.concurrent.Tim
浏览 12提问于2021-05-02得票数 0
回答已采纳
1回答
在流媒体系统中，数据的排序是一个很大的问题。 我们知道，在flink中，对数据的处理是无序的，是使用窗口和水印。 但是，在内部flink中，在操作符之间，如何保证数据的有序性？ flink能否确保先处理高级数据？ 或者在运算符中，数据的顺序打乱了
浏览 265提问于2020-09-22得票数 0
回答已采纳
3回答
我在Flink中做实时流，其中Kafka是消息队列。我正在申请120秒的EventTimeSlidingWindow。滑动1秒。我还在事件时间的每一秒插入水印。
我担心的是，如果元素会在水印之后延迟出现，会发生什么？现在在我的情况下，Flink简单地丢弃了它各自的水印之后的消息。filnk是否提供了任何机制来处理这种延迟消息，比如维护单独的窗口？我也浏览了文档，但我没有弄清楚。
浏览 19提问于2017-01-19得票数 3
回答已采纳
在Apache中，setAutoWatermarkInterval(interval)向下游操作符生成水印，以便他们提前事件时间。
如果水印在指定的时间间隔内没有更改(没有到达事件)，运行时将不会发出任何水印？另一方面，如果在下一个间隔之前到达了一个新事件，则会立即发出新的水印，或者将其排队/等待，直到到达下一个setAutoWatermarkInterval间隔为止。
我很好奇什么是最好的配置AutoWatermarkInterval (特别是对于高速率源而言)：这个值越小，处理时间和事件时间之间的延迟就越小，但是在发送水印时使用更多BW的开销越大。这是真的吗？
另一方面，如果我使用env.
浏览 5提问于2020-08-29得票数 2
回答已采纳
当事件时间水印时，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
回答已采纳
KeyBy和GroupBy在Flink中的异同是什么？如果在仅表程序中使用表/SQL，那么GroupBy是否等同于KeyBy？
浏览 7提问于2020-09-04得票数 1
回答已采纳
flink流具有多个数据流，然后使用org.apache.flink.streaming.api.datastream.DataStream#union方法对这些数据流进行合并。然后，我得到了问题，数据流是混乱的，我不能设置窗口来对数据流中的数据进行排序。 Sorting union of streams to identify user sessions in Apache Flink 我得到了答案，但com.liam.learn.flink.example.union.UnionStreamDemo.SortFunction#onTimer从未被调用过。 环境信息: flink版本1.7
浏览 30提问于2019-01-12得票数 2
我想知道FLINK是否可以支持我的要求，我已经看了很多文章，但我不确定我的情况是否能解决
案例:我有两个输入源。A)事件b)ControlSet事件样本数据为：
event 1-
{
"id" :100
"data" : {
"name" : "abc"
}
}
event 2-
{
"id" :500
"data" : {
"date" : "2020-07-10"
浏览 0提问于2020-07-10得票数 1
我不确定我的Flink应用程序是否真的需要水印。它们什么时候是必要的？
如果我不需要它们，WatermarkStrategy.noWatermarks()的用途是什么？
浏览 4提问于2021-04-21得票数 0
下面是我想在Apache Flink中做的事情： 获取一个输入DataStream<T>，然后按字段x设置关键字，然后执行每分钟滑动一次的15分钟滑动窗口，聚合每个键(x)的结果，然后将所有这些聚合聚合到一个列表中 基本上，如果我有一个输入流，[(a, 1, Time 1), (b, 6, Time 14), (b, 1, Time 12)]，我希望结果是[(a, 1), (b, 7)]，通过操作15分钟的滑动窗口和这个特定的滑动窗口。 这个是可能的吗？
浏览 98提问于2019-02-08得票数 1
回答已采纳
最近我阅读了flink官方文档，了解了一些关于水印的内容。 网址：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/event_time.html 有两张图片说明了flink水印机制，这让我很困惑： flink watermark 第一幅图很容易理解，但在第二幅图中，我想知道如何获得w(11)和w(17)？我们知道，我们可以定义如何在flink作业中生成水印，换句话说，水印是按照一定的规则生成的。那么在第二张图片中生成水印的规则是什么呢？ 我查找了几乎所有不同版本flink的官方文档，它们都使用相同的图片。这让我很
浏览 28提问于2021-03-30得票数 0
我使用的是Flink和一个动态源和事件时间键控窗口。应用程序将监听实时数据流，开窗口(事件时间窗口)并处理每个键控流。我还有另一个用例，在这个用例中，我还需要能够支持回填某些密钥流的旧数据(这些将是带有event-time 考虑到我使用的是水印，这可能是一个问题，因为Flink不支持每个密钥的水印。因此，任何用于回填的键控流最终都将被忽略，因为此流的事件时间将是实况流维护的应用程序水印。
我也问过其他类似的问题，但没有找到可能的方法。以下是我正在考虑的可能方法，但仍有一些悬而未决的问题。
可能的方法-1
(i)备存一份专供回填用途的申请书副本。回填工作很少发生(一个月几次)。发
浏览 18提问于2021-11-01得票数 1
从Flink 1.5发布公告中，我们知道Flink现在支持“广播状态”，并描述了“广播状态为Flink的CEP库的”动态模式“特性的实现打开了障碍。”
这是否意味着目前我们可以使用“广播状态”来实现“动态模式”而不需要Flink CEP？另外，我不知道在有或没有广播状态的情况下实现Flink CEP的“动态模式”有什么区别？如果有人能给出一个用代码解释区别的例子，我会很感激的。
=============
测试广播数据流的更新.使用键控-datastream的操作员广播()
在Flink 1.4.2中进行测试后，我发现广播数据流(由旧的操作程序广播())可以与键控datastream连接，下面
浏览 1提问于2018-05-26得票数 0
回答已采纳
我有一个包含表T1(id，name，age)和T2(id，subject)的数据库。Flink以事件流的形式接收来自数据库的所有更新，使用类似于的内容。这些表是相互关联的，所需数据可以通过连接 T1和T2 on id来提取。目前数据库的全部状态存储在Flink MapState中，其中id作为密钥。现在的问题是，我需要基于名称从T1中选择行，而无需使用id。我似乎需要在T1(name)上建立一个索引，以使它更快。是否有任何方法可以自动索引它，而不必手动为每个表创建索引。这样做的推荐方法是什么？我知道，但我需要对表的更新提供支持。顺便说一下，我在Scala中使用了Flink。如有任何建议或建议，
浏览 0提问于2018-11-22得票数 0
回答已采纳
我在CheckpointedFuntion中使用运算符状态，但是在初始化MapState时遇到了NullPointerException：
public void initializeState(FunctionInitializationContext context) throws Exception {
MapStateDescriptor<Long, Long> descriptor
= new MapStateDescriptor<>(
"state",
TypeInfor
浏览 0提问于2018-03-18得票数 0
1回答
我正在使用事件时间键控窗口的flink。看起来有些窗口没有发射出去。
是否为每个密钥单独添加水印？
例如，如果我的key是( id，type)，而一对特定的id和type没有被摄取到源文件中，那么它们的特定窗口水印是否不会前进？如果是这种情况，我如何确保我所有的keyd窗口在一段时间后都会被逐出？(我们有许多密钥，因此不能为每个密钥定期发送虚拟消息)。
我将非常感谢任何人的帮助
浏览 1提问于2020-04-29得票数 0
我对Apache Flink中的带标点的水印和周期性的水印有点困惑。 假设我有一个包含传入数据的DataStream，数据流类型所属的POJO的时间戳字段始终按升序排列。 所以它应该是类似于[{id: 1, ts: 12}, {id: 2, ts: 13}, ... , {id: 5, ts: 233445}]的东西 我想知道在这种情况下使用哪种类型的水印分配器。我应该使用AscendingTimestampExtractor..还是创建一个自定义的punctuated？
浏览 12提问于2019-02-20得票数 0
我只是想了解在Flink中使用CoProcessFunction的用例。举例说明会帮助我更好地理解这个概念。
浏览 21提问于2019-07-28得票数 1
回答已采纳
TL;DR:目前保证Flink中事件时间顺序的最佳解决方案是什么？
我在Kafka 2.2.1中使用Flink 1.8.0。我需要通过事件时间戳保证事件的正确顺序。我每1s生成一次周期性水印。我将FlinkKafkaConsumer与AscendingTimestampExtractor结合使用：
val rawConsumer = new FlinkKafkaConsumer[T](topicName, deserializationSchema, kafkaConsumerConfig)
.assignTimestampsAndWatermarks(new AscendingTimes
浏览 4提问于2019-10-24得票数 2
回答已采纳
我正在为一个大容量流用例(每秒数以千万计的事件)开发一个DataStream-based Flink应用程序。这些数据是从卡夫卡主题中消耗的，并且已经根据特定的键进行了切分。我的意图是在Flink端创建特定于键的状态，以运行自定义分析。我无法解决的主要问题是，如何创建键控状态，而不对keyBy()强加的传入数据进行重新配置。
我可以保证Flink作业的最大并行性将小于或等于源Kafka主题中的分区数，因此逻辑上没有必要进行洗牌。 StackOverflow问题的答案表明，可能有可能以一种符合Flink预期的方式将数据写入Kafka，然后使用reinterpretAsKeyedStream()。
浏览 4提问于2022-08-08得票数 2
回答已采纳
1回答
我正在使用Flink 1.3.2和scala构建一个流媒体应用程序，我的Flink应用程序将监控一个文件夹，并将新文件流式传输到管道中。文件中的每个记录都有一个关联的时间戳。我想使用这个时间戳作为事件时间，并使用AssignerWithPeriodicWatermarks[T]构建水印，我的水印生成器如下所示：
class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[Activity] {
val maxTimeLag = 6 * 3600000L // 6 hours
override de
浏览 12提问于2018-02-02得票数 0
回答已采纳
1回答
我正在建立一个从Kafka到HDFS的flink管道。我想在addSink()步骤之后处理这些元素。这是因为我希望设置触发器文件，指示为某个分区/小时编写数据(到接收器)已经完成。如何才能做到这一点？目前我正在使用水桶水槽。
DataStream messageStream = env .addSource(flinkKafkaConsumer011)；
//将消息流转换为keyedStream的一些聚合
KeyedStream.addSink(汇)；
//如何在3点之后处理元素？
浏览 0提问于2019-01-08得票数 1
据我理解，Flink中的运算符有源运算符、变换运算符等。对于Flink中的运算符，我的理解是否正确？
在操作符状态中，Flink是为每个作业/任务维护每个操作符的状态(map()、or ()等)还是维护一个完整作业/任务的状态？另外，如果我的作业提交了多个并行性，那么每个槽会有自己的状态吗？
假设，我提交了两个作业，它们是键流，两个作业都有相同的键，比如“颜色”，但这两个任务完全不同。Flink是否会维持两个不同的州，或者它将维持两个工作的一个州。
浏览 0提问于2019-07-22得票数 2
回答已采纳
我们使用flink-cep作为一个独立的库来查找事件列表中的模式。
鉴于下列事件清单：
val patientKey = "patient"
val hrKey = "hr"
// Event
val p1e1 = Event("hr", mapOf(patientKey to 1, hr to 1))
val p1e2 = Event("hr", mapOf(patientKey to 1, hr to 2))
val p2e1 = Event("hr", mapOf(patientKey to 2, hr to
浏览 9提问于2022-10-17得票数 0
2回答
我读到，"..The排序运算符必须缓冲它收到的所有元素。然后，当它接收到水印时，它可以对时间戳低于水印的所有元素进行排序，并按排序的顺序发出它们。这是正确的，因为水印发出的信号是，不会有更多与排序的元素混合在一起的元素到达...“- https://cwiki.apache.org/confluence/display/FLINK/Time+and+Order+in+Streams 因此，水印似乎作为信号传递给下面的操作员，以便开始处理。我想，这也是触发器的作用。这两者有什么不同？
浏览 26提问于2019-03-22得票数 2
1回答
如果我想在Flink中分割一个流，最好的方法是什么？
我可以使用进程函数并使用侧输出来拆分流。水印是否与元素一起传递到侧输出，以便每个侧输出中的数据可以向下传递给其他窗口操作符？
或者，我应该只使用多个filter()操作将一个流过滤到每个包含元素子集的多个流中吗？在这种情况下如何处理水印？所有的水印是否都传递给所有过滤过的流？
如果两者都是可能的，那么哪个是首选(哪个性能更好)？或者，是否有比上述任何一种选择更好的方法？
浏览 3提问于2020-02-24得票数 0
回答已采纳
如何在Flink流中实现反馈循环。换句话说，我如何才能编写一个也是SinkFunction的SourceFunction？
DataStream<Control> controlSignal = ..
DataStream<Data> dataStream = ...
DataStream<Output> outputStream = controlSignal
.connect(dataSignal)
.flapMap(FlatMapFunction)
现在，我希望将outputStream发送到接收器，并根据一些逻辑将数据发送到ControlStre
浏览 0提问于2018-05-17得票数 1
回答已采纳
flink office会话窗口介绍
...a session window operator为每个到达的记录创建一个新窗口，如果窗口之间的距离比定义的间隔更近，则将它们合并在一起。为了能够合并，会话窗口操作符需要合并触发器和合并窗口函数，...
我有一个关于会话窗口合并的问题，如下所示
在这种情况下，两个流合并成一个运算符，user1比user2更早到达，并且user1的第一个窗口与user2的第一个窗口的时间间隔大于GAP，会话窗口如何合并user1和user2的第一个窗口？
在事件时间中，以下哪种情况？图像的上面时间是事件时间。
案例1：
第一个会话窗口: user
浏览 31提问于2020-08-03得票数 0
我有一个流收集和处理的Flink应用程序。该应用程序由Kafka作为事件源、协同映射函数和Kafka接收器组成，如下图所示：
我想用CSV文件中的一些字段丰富Kafka输入的流。我还希望定期刷新配置磁盘位置的csv文件。
潜在的解决方案:广播丰富的数据，并连接来自卡夫卡的事件流，基于某种逻辑。
此外，在配置的间隔之后，继续从文件源重新加载数据。
我对这个设计有以下的关注：
是否可以暂停对Kafka主流的处理，直到使用广播流从文件源加载所有数据为止。
是否有更好的设计来使用CSV文件实现内存中的查找？
浏览 1提问于2022-02-19得票数 0
1回答
如果我连接Flink中的键流和非键流会发生什么？例如；
val stream1 = source.keyBy(_.customerId)
val stream2 = source2.map(operation()...)
stream1.connect(stream2).process(processData).addSink(kafkaSink)
这里的情况是，我试图将一组规则(无键流，stream2)应用到另一个流(stream1)，我通常在广播规则流，但是广播的大小越来越大。所以，我只想知道，如果我不广播其中一个流，并将它们按并行性、状态等连接起来，会发生什么呢？
浏览 8提问于2022-08-07得票数 0
我的问题是，我不能在ORDER BY操作之后使用JOIN子句。为了重现这个问题，
CREATE TABLE stack (
id INT PRIMARY KEY,
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '1' SECONDS
) WITH (
'connector' = 'datagen',
'rows-per-second' = '5',
'fields.id.kind'='sequence
浏览 6提问于2022-08-19得票数 1
回答已采纳
1回答
我目前正在研究Flink上的可伸缩性。从1.2.0版本开始，引入了动态重标度。我正在寻找一个长时间运行的任务，它从Kafka源读取数据。
关于动态重标度的问题。
要扩展我的flink应用程序，例如:添加新的任务管理器，必须重新启动作业/纱线会话才能使用新添加的资源？
我认为可以编写Yarn客户端来部署新的任务管理器并使其与作业管理器对话，这在现有的flink纱线客户端应用程序中已经可用了吗？
请原谅，如果这些问题太基本，我确实看过这些文件，我不得不承认，最近我还没有把这些概念和一些测试部署放在一起。
浏览 1提问于2017-04-18得票数 5
回答已采纳
我们正在构建一个有两个流的应用程序：
大容量的消息流
一个大的静态流(起源于我们周围的一些拼板文件)，我们将其输入Flink，只为了将数据集保存到保存状态。
为了获得共享状态，我们希望将这两个流连接起来，以便第一流可以使用第二状态进行富集。
每天大约，地板文件(第二流的来源)被更新，这将需要我们清除第二流的状态并重建它(可能需要大约2分钟)。
问题是，在此进程运行时，我们能否阻止/延迟来自第一流的消息？
谢谢。
浏览 2提问于2019-01-28得票数 0
回答已采纳
1回答
我们计划使用Apache进行巨大的IOT设置。客户将向我们发送某种结构化的传感器数据(如sensor_id、sensor_type、sensor_value、时间戳)。当每个客户发送这些数据时，我们无法控制，很可能是实时的，但我们无法保证。我们将所有事件存储在RabbitMQ/Kafka中。更新:我们可以假设每个传感器的事件都是有序的。
在开始实现可能的流管道之前，我们对以下挑战的解决方案感兴趣：
多窗口聚类
我们把所有原始传感器数据存储到卡桑德拉。此外，我们希望通过sensor_id在多个时间窗口(例如15 sek、1 min、15 min、1小时、1天)上对传感器数据进行汇总。
浏览 4提问于2016-11-07得票数 4
2回答
我们正在接收来自no的事件。因此，到达我们的Flink拓扑(通过Kafka)的数据将是无序的。
我们在Flink拓扑中创建1分钟的事件时间窗口，并在源操作符处生成事件时间水印(当前事件时间-某个阈值(30秒))。
如果少数事件在设置的阈值之后到达，则这些事件将被忽略(在我们的情况下这是可以的，因为属于该分钟的大多数事件都已经到达并在相应的窗口中得到处理)。
现在，问题是，如果程序崩溃(无论是什么原因)，然后再次从上一个成功的检查点恢复，无序到达的事件将触发过去(已经处理的)窗口的执行(该窗口中只有极小的事件)覆盖了prev的结果。该窗口的计算。
如果Flink具有检查点事件时间水印，则不会发生
浏览 28提问于2018-03-02得票数 5
据我所知，处理Kafka流的状态在内存、磁盘或Kafka主题中都是本地的，因为所有的输入数据都来自一个分区，其中所有的消息都是由一个定义的值键控的。大多数情况下，计算可以在不知道其他处理器的状态的情况下完成。如果是这样的话，您有另一个Streams实例来计算结果。如图所示：
Flink到底在哪里存储它的状态？Flink是否也可以在本地存储状态，或者总是将它们发布到所有实例(任务)？是否可以将Flink配置为将状态存储在Kafka代理中？
浏览 0提问于2019-02-10得票数 3
2回答
我试图在Flink中的KeyedStream上执行地图操作：
stream.map(new JsonToMessageObjectMapper())
.keyBy("keyfield")
.map(new MessageProcessorStateful())
JsonToObjectMapper操作符的输出是MessageObject类的POJO，它有一个字符串字段'keyfield‘。然后在这个字段上输入流。
MessageProcessorStateful是一个像这样的RichMapF
浏览 0提问于2018-02-01得票数 4
回答已采纳
我需要让一个AsyncIO富函数根据最新的规则集执行调用。对于像map这样的操作，我能够通过关注这篇博客文章来处理具有丰富功能的BroadcastConnectedStream：https://flink.apache.org/2019/06/26/broadcast-state.html 但是，创建AsyncIO函数需要DataStream作为输入，而BroadcastConnectedStream不需要(https://ci.apache.org/projects/flink/flink-docs-stable/api/java/index.html?org/apache/flink/s
浏览 29提问于2020-06-19得票数 0
回答已采纳
在卡夫卡主题中，我有{“主机名”：“abc.com”、“数据”：{}}这样的事件。这些事件是json对象。我的flink-kafka消费者正在访问这些事件。现在，在flink中，我可以根据主机名对这些事件进行分组吗?在每个组中，我可以检查模式吗？
浏览 0提问于2018-12-07得票数 0
回答已采纳
我是新来的Flink，我试着做一些练习，当我使用窗口与卡夫卡连接器，我遇到了一个问题。
代码如下：
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
.....
// a KeyedStream with name as the key
val eventsStream: KeyedStream[CustomedObject, String] = env
.addSource[CustomedObject](source.getSource)
.keyBy
浏览 3提问于2017-11-27得票数 0
回答已采纳
我正在构建一个Flink流系统，它可以同时处理实时数据和历史数据。所有数据都来自同一个源，然后在split中进入历史和动态。实时数据得到时间戳和水印，而历史数据则按顺序接收.在活动流被加窗后，两个流被合并并流到相同的处理管道中。
如果EventTime流环境中的所有记录都需要时间戳，或者Flink甚至可以同时处理这种实时和历史数据，我就找不到了。这是一个可行的方法，还是会造成我经验不足而看不到的问题？数据的顺序会受到什么影响？
我们有这样的设置，允许我们进行部分回填。每个流都是由一个id键决定的，我们发送历史数据来替换一个id的观测数据，同时不影响其他id的实时处理。
这是作业图：
浏览 2提问于2018-12-13得票数 0
回答已采纳
我从四条动态流中读取数据。每个流中的数据是不同的数据类型。在读取所有四个流之后，我分配时间戳和水印，并从每个流中聚合数据。四个聚合的结果都是使用相同的泛型对象输出的。我希望合并来自四个流的结果，这样我就可以将统一的流发送到一个ProcessFunction。这基本上允许我像使用ProcessFunction一样使用CoProcessFunction，但我可以处理来自两个以上流的数据(在这种情况下，ProcessFunction将接收来自所有四个单独流的聚合)。
然而，我担心的是，这可能与水印不太好。如果一个流需要更长的时间来处理，或者在某种程度上是落后的，那么如果所有的水印都在联合中向前传递，
浏览 1提问于2020-02-24得票数 2
回答已采纳
我正在用DataStream应用程序接口在Flink中做时间序列数据分析。每50个事件(数据收入)，我想用它们进行转换，保存结果，并将其与下50个事件的转换结果进行比较。以此类推。我的问题是，在进行转换后，在Flink中有没有一种方法来保存结果，并查询下一个50个事件的下一个转换，以及之前的结果？
浏览 0提问于2020-03-18得票数 0
1回答
根据我对Flink的理解，介绍了基于键(键组)的并行性。然而，假设一个人有一条巨大的没有钥匙的小溪，并且希望并行地完成这项工作，那么实现这一目标的最佳方法是什么？
如果流中有一些字段，您可能会考虑按某个字段按键，但是这并不保证工作负载将得到适当的平衡。例如，因为该字段中的一个值可能发生在90%的消息中。因此，我的问题是：
如何在Flink中执行良好平衡的并行性，而无需事先了解流中的内容
我可以想到的一个潜在的解决方案是为每条消息分配一个随机数(如果您想要并行性为3，或者如果您希望并行性更灵活，则为1-1000 )。但是，我想知道这是否是推荐的方法，因为它感觉不太优雅。
浏览 0提问于2020-06-28得票数 2
回答已采纳
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频
腾讯云开发者
