开发者社区> 问答> 正文

flink CDC使用join的时候表更新如何反馈到sink table

Meta

Source:【①群】Apache Flink China社区
Time:2021/11/10 10:44
Proposer:王杰
Status:Unresovled

Content

  • 王杰
    flink cdc 做数据打宽,insert into C select * from A JION B 在第一次同步到数据后,T+1场景单独更新A或者B,C更新不了全部列的数据,这种情况有什么最佳实践吗?

展开
收起
zyxing 2022-03-22 16:05:20 2700 0
1 条回答
写回答
取消 提交回答
  • flinkEnv.env() // A 流 .addSource(new SourceFunction() { @Override public void run(SourceContext ctx) throws Exception {

        }
    
        @Override
        public void cancel() {
    
        }
    })
    // B 流
    .join(flinkEnv.env().addSource(new SourceFunction<Object>() {
        @Override
        public void run(SourceContext<Object> ctx) throws Exception {
            
        }
    
        @Override
        public void cancel() {
    
        }
    }))
    // A 流的 keyby 条件
    .where(new KeySelector<Object, Object>() {
        @Override
        public Object getKey(Object value) throws Exception {
            return null;
        }
    })
    // B 流的 keyby 条件
    .equalTo(new KeySelector<Object, Object>() {
        @Override
        public Object getKey(Object value) throws Exception {
            return null;
        }
    })
    // 开窗口
    .window(TumblingEventTimeWindows.of(Time.seconds(60)))
    // 窗口中关联到的数据的处理逻辑
    .apply(new JoinFunction<Object, Object, Object>() {
        @Override
        public Object join(Object first, Object second) throws Exception {
            return null;
        }
    });
    

    2022-05-23 12:42:56
    赞同 展开评论 打赏
问答排行榜
最热
最新

相关电子书

更多
Flink CDC Meetup PPT - 龚中强 立即下载
Flink CDC Meetup PPT - 王赫 立即下载
Flink CDC Meetup PPT - 覃立辉 立即下载