zoukankan      html  css  js  c++  java
  • Flink使用SideOutPut替换Split实现分流

    以前的数据分析项目(版本1.4.2),对从Kafka读取的原始数据流,调用split接口实现分流.
    新项目决定使用Flink 1.7.2,使用split接口进行分流的时候,发现接口被标记为depracted(后续可能会被移除).
    搜索相关文档,发现新版本Flink中推荐使用带外数据进行分流.

    预先建立OutputTag实例(LogEntity是从kafka读取的日志实例类).

    private static final OutputTag<LogEntity> APP_LOG_TAG = new OutputTag<>("appLog", TypeInformation.of(LogEntity.class));
    private static final OutputTag<LogEntity> ANALYZE_METRIC_TAG = new OutputTag<>("analyzeMetricLog", TypeInformation.of(LogEntity.class));

    kafka读取的原始数据,通过process接口,打上相应标记.

    private static SingleOutputStreamOperator<LogEntity> sideOutStream(DataStream<LogEntity> rawLogStream) {
            return rawLogStream
                    .process(new ProcessFunction<LogEntity, LogEntity>() {
                        @Override
                        public void processElement(LogEntity entity, Context ctx, Collector<LogEntity> out) throws Exception {
                            // 根据日志等级,给对象打上不同的标记
                            if (entity.getLevel().equals(ANALYZE_LOG_LEVEL)) {
                                ctx.output(ANALYZE_METRIC_TAG, entity);
                            } else {
                                ctx.output(APP_LOG_TAG, entity);
                            }
                        }
                    })
                    .name("RawLogEntitySplitStream");
        }
    
        // 调用函数,对原始数据流中的对象进行标记
        SingleOutputStreamOperator<LogEntity> sideOutLogStream = sideOutStream(rawLogStream);
        // 根据标记,获取不同的数据流,以便后续进行进一步分析
        DataStream<LogEntity> appLogStream = sideOutLogStream.getSideOutput(APP_LOG_TAG);
        DataStream<LogEntity> rawAnalyzeMetricLogStream = sideOutLogStream.getSideOutput(ANALYZE_METRIC_TAG);

    通过以上步骤,就实现了数据流的切分.

    参考:https://copyfuture.com/blogs-details/20190929214201974igwkcfcfuc2cph3

  • 相关阅读:
    MOSS发生“未能转换部分或所有标识引用”错误的处理
    Excel嵌入网页(二) 转载
    GridView的PagerTemplate分页
    深入探讨:如何理解.Net的三层架构(转)
    IIS报Service Unavailable错的解决方案
    浅谈C#托管程序中的资源释放问题(转)
    C#程序节省内存
    网页中嵌入Excel控件(补充)
    解决ajax的status=0问题
    翻开我的博客之旅
  • 原文地址:https://www.cnblogs.com/yoyowin/p/13402188.html
Copyright © 2011-2022 走看看