问题背景

Flink 的一种常见应用场景是 ETL（抽取、转换、加载）管道任务。从一个或多个数据源获取数据，进行一些转换操作和信息补充，将结果存储起来。整个了流程如下图所示：

这个链路是非常场景的一个数据上报链路。常见的做法是将转换的配置用一些规则表达式的脚本语言来编写，然后统一存入到数据库中（MySQL/MongoDB都可以），接着当Flink启动后，从DB加载配置，实现规则读取步骤。当新的Kafka数据抵达后，可以从中提取对应的脚本进行转换即可。逻辑链路如下图所示：

这样的设计思路其实是正确的。但是这里面会存在一个问题，就是如何从DB中加载我们的配置。

广播流

在业内比较常见的思路是将配置都用一个广播流加载进来，然后定期刷新的时候也是通过广播流广播给到各个算子。

但是这种思路在遇到高并发场景的时候，可能会在Flink首次加载配置的时候遇到广播流还未加载，数据流却先抵达的场景。

在flink官方中文邮箱列表中找到了一个合理的解决上述问题的思路，在处理数据流的方法中使用ListState对数据进行“缓冲”，等到配置信息加载完毕后再对“缓冲”的数据进行处理，下面是我自己的实现。首先是Flink加载DataStream的代码片段：


import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @author idea
 * @description flink批处理案例
 */
public class FlinkKafkaETLDemo {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String broker = "localhost:9092";
        String groupId = "flink-group";
        String topic = "test-topic";
     
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        BroadcastStream> mockBroadcastStream = env.addSource(new ConfigRichFunction()).name("s_condig_rich_function")
                .setParallelism(1).broadcast(ConfigProcessFunction.getConfig());


        for (int i = 0; i < 10; i++) {
         /** 模拟10个不同的topic数据流读取，然后交给不同的处理规则进行转阿虎呢**/
            String topicName = topic + "-" + i;
            FlinkKafkaConsumer kafkaConsumer = new FlinkKafkaConsumer<>(topicName, new SimpleStringSchema(), properties);
            DataStream testDataStream = env.addSource(kafkaConsumer)
                    .name("s_" + topicName)
                    .keyBy(new KeySelector() {
                        @Override
                        public Object getKey(String s) throws Exception {
                            return s;
                        }
                    })
                    //connect流和process流join在一起
                    .connect(mockBroadcastStream)
                    .process(new DynamicProcessFunction(topicName));
            testDataStream.print();
        }
        try {
            env.execute("batch-kafka-test");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

接着是自定义的广播流：

import  org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import  java.io.Serializable;
import  java.util.ArrayList;
import  java.util.HashMap;
import  java.util.List;
import  java.util.Map;
import  java.util.concurrent.TimeUnit;

/** 
* @author idea 
* @description 读取配置到本地内存的广播流
 */
public  class  ConfigRichFunction  extends  RichSourceFunction < List < Map >>  implements  Serializable  {   
    private  volatile  boolean  running =  true;   @
    Override    public  void  run(SourceContext  sourceContext) throws  Exception  {     
        while  (running) {        /**模拟从DB加载配置到本地内存**/        
            System.out.println("load config");      
            sourceContext.collect(this.queryDataFromMock());       
            TimeUnit.SECONDS.sleep(5);    
        }  
    }   @
    Override    public  void  cancel() {     
        System.out.println("close rich config function");    
        running =  false;  
    }   
    private  List < Map >  queryDataFromMock() {     
        List < Map > lists =  new  ArrayList < > ();      /** 模拟不同的topic对应的配置不同 **/      
        for  (int i =  0; i <  10; i++) {       
            Map < String,  String > mapRecord =  new  HashMap < > ();      
            mapRecord.put("test-topic-"  + i,  "item-1");      
            lists.add(mapRecord);    
        }     
        try  {       
            TimeUnit.SECONDS.sleep(3);    
        } 
        catch  (InterruptedException  e) {       
            throw  new  RuntimeException(e);    
        }     
        return  lists;  
    }
}

针对这个配置加载流，我们可以自定义一个抽象父类，用于做基础的定义：

import  org.apache.flink.api.common.state.MapStateDescriptor;
import  org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import  org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import  org.apache.flink.api.common.typeinfo.TypeHint;
import  org.apache.flink.api.common.typeinfo.TypeInformation;
import  org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import  org.apache.flink.util.Collector;
import  java.util.List;
import  java.util.Map;

/** 
* @author idea 
* @description 读取转换脚本，并且广播给自定义的processfunction 
*/
public  abstract  class  ConfigProcessFunction < KS, IN, OUT >  extends  KeyedBroadcastProcessFunction < KS, IN,  List < Map > , OUT > {   
        private  final  static  String  PROP_CONFIG  =  "prop_config";   
        public  static  MapStateDescriptor < String,  List < Map >>  MOCK_CONFIG_DESCRIPTOR  =  new  MapStateDescriptor < > (       
            "mockConfig",        
            BasicTypeInfo.STRING_TYPE_INFO,        
            TypeInformation.of(new  TypeHint < List < Map >> () {      })
            );   
        public  static  MapStateDescriptor < String,  List < Map >>  getConfig() {     
            return  MOCK_CONFIG_DESCRIPTOR;  
        }   
        @Override    
        public  void  processElement(IN  content,  KeyedBroadcastProcessFunction < KS, IN,  List < Map > ,  OUT > .ReadOnlyContext  ctx,  Collector < OUT > collector) throws  Exception  {     
        System.out.println("处理"  + content);     
        ReadOnlyBroadcastState < String,  List < Map >> mockPropBroadcastState = ctx.getBroadcastState(MOCK_CONFIG_DESCRIPTOR);     
        List < Map > mockConfigs = mockPropBroadcastState.get(PROP_CONFIG);     
        process(content, ctx, collector, mockConfigs);  
    }

        /**     * 进行规则转换处理     
        * @param content     
        * @param ctx     
        * @param collector     
        * @param mockConfigList     
        */    
        public  abstract  void  process(IN  content,  ReadOnlyContext  ctx,  Collector < OUT > collector,  List < Map > mockConfigList) throws  Exception;

       @Override    
       public  void  processBroadcastElement(List < Map > jsonObjects,  KeyedBroadcastProcessFunction < KS,  IN,  List < Map > ,  OUT > .Context  context,  Collector < OUT > collector) throws  Exception  {     
        System.out.println("放入数据");    
        context.getBroadcastState(MOCK_CONFIG_DESCRIPTOR).put(PROP_CONFIG, jsonObjects);  
    }
}

到这里你可能感觉代码还比较正常。下边是如何基于设置ListState缓存提前抵达的数据流。

import  org.apache.flink.api.common.state.*;
import  org.apache.flink.api.common.typeinfo.TypeHint;
import  org.apache.flink.api.common.typeinfo.TypeInformation;
import  org.apache.flink.configuration.Configuration;
import  org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import  org.apache.flink.util.Collector;
import  java.util.List;
import  java.util.Map;
/** 
* @author idea 
* @description 动态识别不同的topic，执行不同的转换脚本逻辑 
*/
public  class  DynamicProcessFunction  extends  ConfigProcessFunction < String, String,  String > {    /** 配置信息未加载完毕时先缓存数据流元素 **/    
    private  transient  ListState < String > payloads;    /** 未设置定时输出缓存流元素的定时器，默认未设置 **/    
    private  transient  ValueState < Boolean > setTimer;   
    private  String  topic;   
    public  DynamicProcessFunction(String  topic) {     
        this.topic  = topic;  
    }   
       @Override   
      public  void  open(Configuration  parameters) throws  Exception  {     
        super.open(parameters);     
        System.out.println("开启处理流"  + topic);    
        payloads =  getRuntimeContext().getListState(new  ListStateDescriptor < > (FlinkConstants.LIST_STA,  TypeInformation.of(new  TypeHint < String > () {    })));    
        setTimer =  getRuntimeContext().getState(new  ValueStateDescriptor < > (FlinkConstants.TIMER_STA,  TypeInformation.of(new  TypeHint < Boolean > () {    })));  
    }   
      @Override    
      public  void  process(String  content,  KeyedBroadcastProcessFunction < String, String,  List < Map > ,  String > .ReadOnlyContext  ctx,  Collector < String > collector,  List < Map > mockConfigList) throws  Exception  {     
        if  (isBroadcastStreamReady(mockConfigList)) {       
            if  (setTimer !=  null  && setTimer.value()) {         
                Iterable < String > iterable = payloads.get();         
                for  (String  cacheMsg: iterable) {           
                    System.out.println("处理之前缓存的数据内容"  + cacheMsg);          
                    collector.collect(cacheMsg);        
                }        
                payloads.clear();        
                setTimer.update(false);      
            } 
            else  {        
                collector.collect(content);      
            }    
        } 
        else  {       
            System.out.println("广播流未抵达异常："  + content);       
            try  {        
                payloads.add(content);          /** 只需要一个定时器就行了，到时执行一次就可以把缓存的所有数据处理掉 **/          
                if  (null  == setTimer.value() || !setTimer.value()) {          
                    long nextProcessTime = ctx.timerService().currentProcessingTime() + 10000;          
                    ctx.timerService().registerProcessingTimeTimer(nextProcessTime);          
                    setTimer.update(true);        
                }      
            } 
            catch  (Exception  e) {         
                throw  new  RuntimeException(e);      
            }    
        }  
    }

       @Override    
       public  void  close() throws  Exception  {     
        super.close();     
        System.out.println("关闭处理流"  + topic);  
    }

       @Override    
       public  void  onTimer(long timestamp,  KeyedBroadcastProcessFunction < String,  String,  List < Map > ,  String > .OnTimerContext  ctx,  Collector < String > out) throws  Exception  {     
        super.onTimer(timestamp, ctx, out);     
        List < Map > mockConfigs = ctx.getBroadcastState(ConfigProcessFunction.getConfig()).get(FlinkConstants.PROP_CONFIG);     
        System.out.println("之前的配置：" + mockConfigs);     
        for  (String  element: payloads.get()) {      
            out.collect(element);       
            System.out.println("处理之前堆积的异常数据2:" + element);    
        }    
        payloads.clear();      /** 执行完任务删除定时器 **/     
        ctx.timerService().deleteProcessingTimeTimer(timestamp);    
        setTimer.update(false);  
    }   
    private  boolean  isBroadcastStreamReady(List < Map > mockConfigList) {     
        return  mockConfigList !=  null;  
    }
}

这段代码我大致说下思路：

程序启动，数据流元素和配置流元素开始加载，不分先后；
当接收到数据流元素时，首先判断配置流元素是否加载完毕（见isBroadcastStreamReady方法）。若未加载完毕，会把数据缓存起来，因为流处理是数据驱动的，数据不来，缓存的数据就不会被处理。所以为了解决这种情况，需要创建一个定时器，默认10秒后自动把缓存的数据处理掉（之所以设置10秒，是因为广播流每间隔5秒读取一次配置，设置10秒是考虑了两倍的延迟周期）；
当ListState缓存了一定的数据，如果下一个数据流元素来的时候发现配置已经加载完毕了，则首先把缓存的数据处理掉，同时清除定时器TImer的标志等，最后再是处理收到的数据，后续流程就照此正常流转即可了；
最后我把整份代码的数据流转图绘制了出来，方便大家更好的理解。

小总

本文提供的是ETL在Flink应用中落地的一种解决思路，这样设计能够尽可能避免在使用Flink时候出现因为流聚合顺序的问题，导致数据转换丢失异常情况发生，更好的提升我们Transform这一环节的准确性。希望能对大家有所帮助。