flink-user-zh mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From 徐涛 <xutao_u...@163.com>
Subject Re:Re: Re: Flink如何实现Job间的协同联系?
Date Mon, 01 Jul 2019 06:31:53 GMT
Hi 军长,
      是的,我的意思正是这样。我的希望是以Kafka作为串联多个Flink Job运行的媒介,但是又觉得Kafka这种介质对于错误处理不够友好。
      如果是对于跑批的Hive表,那么上游跑完以后,下游能看到的数据就是上游覆盖后的数据,整个系统非常的干净。代码也不用修改。
      但对于流式应用来说,如果上游更新逻辑后重跑,Kafka中的旧有数据不会被清空,那么会导致下游重跑后不知道怎么去从Kafka中精确筛选出上游新逻辑产生的数据进行消费。
      您推荐的方法很新颖,即关闭时上游先关,下游后关;启动时下游先启动,上游后启动,而下游的Job使用Kafka
latest取上游数据,这样对于简单的场景确实可以解决问题,非常感谢!


      同时对于复杂场景,使用这种模式我也进行了思考。如果对于如下类型的拓扑结构
      job1   --->    job2  --->   job3
                          /\
                           |
      job5   --->   job4
      如果job1进行了逻辑修改,是否应该这样的步骤:
      1. 关闭job1, job2, job3
      2. 启动job3,  使用latest拉取job2产生的Kafka topic结果
      3. 启动job2,使用latest拉取job1产生的Kafka topic结果;使用earliest拉取job4产生的Kafka
topic结果
      4. 启动job1
这里的问题在于---job5,job4没有更改,因此不需要重跑,所以在启动job2时需要对job1和job4产生的topic采用不同的模式来追数据,一个是latest一个是earliest。
当然不一定要用earliest和latest,我这里只是举了个例子,但是想说明的是,某个job如果有多个上游的时候,可能需要采用不同的Kafka模式来追数据。


另外不知道在阿里是采用什么模式解决流式任务使用Kafka作为依赖媒介这种场景的呢?也是采用“关闭时上游先关,下游后关;启动时下游先启动,上游后启动”这种模式吗?




谢谢
徐涛

在 2019-06-28 10:01:35,"Hequn Cheng" <chenghequn@gmail.com> 写道:
>Hi 徐涛,
>
>我理解你应该是想上下游匹配起来,而且新的上游产出不能给老下游了。
>
>有一个办法是,把上下游的job同时重启,这样不用新写到其他topic,而且保证结果符合预期。
>具体的操作流程是:停止上游->停止下游->启动下游(从topic当前时间点消费数据)->启动上游开始追数据。
>如果下游job能及时消费上游的产出,当上游job追上当前进度时,下游job输出也是最新的。
>
>Best, Hequn
>
>On Wed, Jun 26, 2019 at 5:40 PM 徐涛 <xutao_ustc@163.com> wrote:
>
>> Hi 军长,
>>     谢谢您的回复。
>>     对于问题“如果上游逻辑更改,重新跑数据,那么可能会存在最开始的那天数据不完整,导致污染下游数据”。我觉得这个问题对于想拿Flink做一些规模稍大的系统(或者说基于Kappa架构的设计)可能都会遇到相同的问题。如果对于跑批的场景,因为上游重跑可以覆盖中间结果,下游可以拿到更新后的数据并进行计算;但是流式计算Job的中间结果落盘于Kafka,而且下游的Job已经累积了一些状态,这个时候上游的计算逻辑如果发生了更改,如果还是写到同一个Kafka
>> topic,那么很难保证下游数据的正确性;如果写到不同的Kafka topic,那么下游的实时任务可能都需要重启,这样操作的负担会很重。
>> 不知道一般对于实时数仓来说是怎样解决这样的问题的?
>>
>> 谢谢!
>> 徐涛
>>
>>
>>
>>
>> 在 2019-06-21 21:21:34,"Hequn Cheng" <chenghequn@gmail.com> 写道:
>> >Hi 徐涛,
>> >
>> >最后四个问题,3和4和业务逻辑关系大些,我大致说下1和2的想法,
>> >
>> >1. 因为每个变化都要发一个message,Kafka中的消息数可能很大。
>> >这个问题,可以借助一些攒批的策略来减少数据量,比如在自定义的sink里加些cache,每1000条刷一次,刷的时候对这1000条按key去重。需要注意的是,攒批需要结合checkpoint一起来考虑,防止丢数据,可以参考下JDBCOutputFormat的实现(做checkpoint的时候需要flush)。此外,blink里面aggregate也有miniBatch的逻辑,也会减少输出的数据量(不过,flink里面还没有miniBatch)。
>> >
>> >2. 下游做幂等的时候,因为要对每个Key做Group by,因此消耗的资源也很大。
>> >这个是的,目前应该没有很好的办法。后期,如果支持了RetractSource,下游job可以不用再做groupBy+last。只需要上游job用RetractSink输出存一份数据到Kafka。
>> >
>> >一些其他问题:
>> >> ① 需要实现一个retract kafka sink
>> >这里应该是需要实现一个upsert kafka sink,目前flink还没法输入retract
message。
>> >
>> >Best,Hequn
>> >
>> >On Wed, Jun 19, 2019 at 1:18 PM 徐涛 <xutao_ustc@163.com> wrote:
>> >
>> >> 大家好,
>> >>
>> >> 我这边想做一个实时数仓项目。但随着Flink间Job的数量越来越多,发现很多Job之间的代码存在大量的重复,且迫切需要一个类似于Batch的中间层解决方案来减少冗余,增加整体的清晰度和层次感。
>> >>     我这边能想到的解决方案是:采用Kafka作为Job之间的联系纽带,例如有两个Job
>> >>     Job_1:   从原始kafka topic TOPIC_ORIG获取数据, 进行一定的业务逻辑处理后,写到另一个kafka
topic,
>> >> TOPIC_JOB_1_SINK 。注意
>> >>                ① 需要实现一个retract kafka sink
>> >>                ② 没有使用kafka exactly-once sink
>> >>                ③ TOPIC_JOB_1_SINK 中的每条记录应该有一个 unique
key.
>> >>                ④ 每个Key相同的记录应该被发往相同的kafka
partition.
>> >>     Job_2:    从TOPIC_JOB_1_SINK读取数据, 接着做幂等(先对唯一Key做group
by取最新),
>> >> 然后运行Job_2的逻辑 , 最后把数据写道最终Sink中(例如es,
hbase, mysql)。 之所以要对unique
>> >> key做一轮幂等处理,因为Job_1可能会失败重试,TOPIC_JOB_1_SINK中可能会有一些重试脏数据。
>> >>
>> >>
>> >> 从整体上看,结构大概如下图所示:
>> >> Job_1Job_2
>> >> -------------------------------------------------------------------------------------
>> >>
>> >>  -----------------------------------------------------------------------------------------------------------------------------------------------------------
>> >> |      TOPIC_ORIG -> Logic_Job_1 -> TOPIC_JOB_1_SINK      |      
    ——>
>> >>    |     TOPIC_JOB_1_SINK -> GROUP_BY_UNIQUE_KEY_GET_LATEST -> Logic_Job_2
>> >> -> FINAL_JOB_2_SINK    |
>> >> -------------------------------------------------------------------------------------
>> >>
>> >>  -----------------------------------------------------------------------------------------------------------------------------------------------------------
>> >>
>> >>
>> >> 即:每个Job往下游发送的数据整体有唯一Key;每个下游需要对上游发来的数据做幂等处理。
>> >> 但是,可能存在的问题有:
>> >> 1. 因为每个变化都要发一个message,Kafka中的消息数可能很大。
>> >> 2. 下游做幂等的时候,因为要对每个Key做Group by,因此消耗的资源也很大。
>> >> 3. 如果上游逻辑更改,重新跑数据,那么可能会存在最开始的那天数据不完整,导致污染下游数据
>> >> 4. 如果上游逻辑更改,重新跑数据,但是某条数据这个时候已经不应该出现了,下游的数据得不到更新
>> >>
>> >>
>> >> 请问大家是如何管理Flink Job之间的依赖或者说血缘关系的?有没有比较好的方案?
>> >> 谢谢大家!
>> >>
>> >>
>> >> 谢谢
>> >> 徐涛
>>
>>
>>
>>
>>
Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message