Re: checkpoint stage size的问题

你好 从附件的web监控看，其实你的整体checkpoint state其实很小（只有20几MB），所以对于这个问题其实有些过度关注了。 关于checkpoint state的变化，需要观察不同operator的情况，可以点开详细页看每个并发的情况。对比operator state和window所使用的keyed state的变化情况。我估计keyed state部分会有些许波动，主要是因为你使用的是RocksDB state backend，其实上传的是rocksDB的sst文件，当register timer时，window state会进行存储，当onTimer时，相关stat

Re:Re: Flink如何实现Job间的协同联系？

Hi 军长， 谢谢您的回复。 对于问题“如果上游逻辑更改，重新跑数据，那么可能会存在最开始的那天数据不完整，导致污染下游数据”。我觉得这个问题对于想拿Flink做一些规模稍大的系统（或者说基于Kappa架构的设计）可能都会遇到相同的问题。如果对于跑批的场景，因为上游重跑可以覆盖中间结果，下游可以拿到更新后的数据并进行计算；但是流式计算Job的中间结果落盘于Kafka，而且下游的Job已经累积了一些状态，这个时候上游的计算逻辑如果发生了更改，如果还是写到同一个Kafka topic，那么很难保证下游数据的正确性；如果写到不同的Kafka topic，那么下游的实时任务可能都

Re: checkpoint stage size的问题

这是web ui的监控 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: checkpoint stage size的问题

-- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: checkpoint stage size的问题

老师你好，首先感谢你在百忙之中回复我。 我这面观察到的现象是，当有数据流入时，每个checkpoint的stage size比上一个checkpoint多几百k左右，只要数据持续流入，这个stage size就一直增长，当没有数据流入时，checkpoint的stage size就维持不变了，再有数据流入时，stage size就在原来基础上继续增长。 数据流： SingleOutputStreamOperator studentSubjectStream = dataStream .filter(new Question2SubjectFilter())

Re: checkpoint stage size的问题

你好 这个问题问得有点稍微宽泛，因为并没有描述你所认为的checkpoint state size越来越大的周期。checkpoint state size变大有几个原因： 1. 上游数据量增大。 2. window设置时间较长，尚未触发，导致window内积攒的数据比较大。 3. window的类型决定了所需要存储的state size较大。 可以参考社区的文档[1] window state的存储空间问题。另外，在上游数据量没有显著变化的时候，若干窗口周期后的checkpoint state size应该是比较稳定的，由于未明确你的观察周期，所以只能给出比较宽

Re: checkpoint stage size的问题

我发现窗口的trigger只进行了fire，并没有进行purge，我不清楚是不是这个原因，或者还是有其他的原因。 -- Sent from: http://apache-flink.147419.n8.nabble.com/

来自小乐的邮件

checkpoint stage size的问题

各位好，我的项目的流计算模型source(kafka)->filter->keyby->window->aggregate->sink(hbase),现在发现window的subtask的checkpoint的stage size越来越大，请问是什么原因啊？ -- Sent from: http://apache-flink.147419.n8.nabble.com/

关于使用Flink建设基于CDC方式的OGG数据湖

一：背景描述现有数据中心基于GP做数仓，但是在OGG数据到GP做贴源ODS过程耗费集群太多资源，导致集群性能瓶颈，故想考虑基于GP + HADOOP的架构来重构数据仓库。 二：思路将入库贴源ODS的工作交由Hadoop生态来完成，数据从OGG实时流入kafka时，都是类似DDL中的"I, U ,D"类型数据, 然后Flink采用动态表方式找出每条主键的最新记录，内存中的动态表将按天写入HDFS, 然后GP通过外部表的方式，将今天的数据加载到GP中与存量做meger更新，这样既完成性能瓶颈的调优， 数据流转图简要如下： 三：问题1、现采用新技术路线，在kafka数据处理后转为toRetr