flink动态业务规则

# Flink动态业务规则的实现 [TOC] ## 一、 需求 在不停止flink任务的情况下，在外部通过配置，来达到修改flink输出特征逻辑的变化、新增和删除。 ## 二、 监控配置变化 ### 2.1 FlinkCDC的实现 直接使用 `flink-connector-mysql-cdc` ### 2.2 Apollo的实现 自定义apollo source，没隔1s输出期间内监控到的apollo配置的变化。 ```java import com.alibaba.fastjson.JSON; import com.ctrip.framework.apollo.Config; import com.ctrip.framework.apollo.ConfigChangeListener; import com.ctrip.framework.apollo.ConfigService; import com.ctrip.framework.apollo.model.ConfigChange; import flink_dynamic_rules.entity.ApolloEvent; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.source.RichSourceFunction; public class ApolloSource extends RichSourceFunction<String> { private Config config; private ApolloEvent event = null; private boolean flag = true; @Override public void open(Configuration parameters) throws Exception { super.open(parameters); // 一般修改一个配置间隔时间几秒，就是说几秒之内会只监听到一个事件 // 如果考虑高并发的情况，把事件放进队列中，在run中取出队列中所有的事件 ConfigChangeListener changeListener = changeEvent -> { for (String key : changeEvent.changedKeys()) { ConfigChange change = changeEvent.getChange(key); event = new ApolloEvent(change.getPropertyName(), change.getOldValue(), change.getNewValue(), change.getChangeType().toString()); } }; config = ConfigService.getAppConfig(); config.addChangeListener(changeListener); } @Override public void run(SourceContext<String> sourceContext) throws Exception { while (flag) { if (event != null) { sourceContext.collect(JSON.toJSONString(event)); event = null; } Thread.sleep(1000); } } @Override public void cancel() { flag = false; } } ``` ## 三、 根据配置的变化修改逻辑 ### 3.1 简单的逻辑 逻辑变化不大，只是一些阈值或者系数发生变化的特征。 把这些阈值和系数看作维表，与数据流连接做特征的计算。 **数据流：** | session_id | score1 | score2 | score3 | | ---------- | ------ | ------ | ------ | | | | | | **配置流：** | a | b | c | d | | ---- | ---- | ---- | ---- | | 3 | 2 | 100 | xxx | **特征计算逻辑：** ```sql select session_id, score1*a + score2*b as feature1, if(score3>c,1,0) as feature2 from connect_stream; ``` ### 3.2 复杂的逻辑 对于计算逻辑比较复杂的，需要用udf计算的特征。上面的方法就不能实现了。可以通过动态加载类，再通过反射执行类的方法，来实现特征的输出。 **具体的实现步骤： **flink_dynamic_rules.app.Demo2 1. 获取apolo/flink cdc配置流 2. 获取数据流 3. 广播配置流 4. 数据量与配置流连接 connectDS 5. connectDS 的BroadcastProcessFunction - 处理广播流 把广播流的数据（类名，jar包路径）添加进mapState中 - 处理数据流 遍历mapState中的数据，通过类加载器加载，再通过反射执行方法 6. 输出 sink