Flink：把状态State全都扒光，远走他乡(二)：广播状态

先用一个场景来入门：我们想象的是一个电商平台的用户操作和模式的实时匹配的情况吧。它获取了所有用户的操作行为数据作为一个用户的操作流。网站的运营团队致力于分析用户的操作，来提高销售额，改善用户体验，并监测和预防恶意行为。要实现了一个流应用程序，用于检测用户事件流中的模式。当然，也可以在代码中把所谓的这种“模式”给写死，但是这样情况是很不理想的对吧，总是要重新部署我们的应用，而且，那样用不到广播状态，而直接只能和UserActions流一起传播，这样相同的广播状态就重复传播，占用资源。
他是这样的

• 定义一个User actions流，用来记录各个用户的操作，比如登陆，加入购物车，付款，注销退出等等。
• 定义一个Patterns流，用来记录平台需要的模式匹配，比如某个模式是记录这个用户在登陆之后马上退出了，或者这个用户加入购物车后马上退出了，总之就是一个动作之后接另一个动作形成一个模式，当然也可以多个动作连起来，不过这里为了简便，就暂时两个动作。

像下面这个样子：

我们可以看到下面这个Patterns是一个模式流，它两个作为一组，也就是两个作为一个Pattern，广播给其他的operator。让广播出去的operator去匹配User Actions。

将Pattern广播出去了：从图中可以看出，是登陆->退出的Pattern广播出去了。这样的话。

然后就是User Actions流根据key来分到每一个对应的分区并行操作。这是多么的好。而且呢，每次一个action到来了，每个分区都会保存上一次的action，联合刚来的action一起组成一个模式和广播的Pattern去匹配，最后再讲刚来的action替代上次的action。

可以看到Key1的操作匹配到了Pattern，然后就collect

但是如果新来一个Pattern的话，就会替代掉原来的那个Pattern，从而就会与新来的Pattern比较了。

代码实现

package flinkjava.State;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;public class broadcastState {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();SingleOutputStreamOperator<Action> actions = env.socketTextStream("127.0.0.1", 9000).map(new MapFunction<String, Action>() {@Overridepublic Action map(String value) throws Exception {String[] dataArray = value.split(",");return new Action(Long.valueOf(dataArray[0].trim()), dataArray[1].trim());}});SingleOutputStreamOperator<Pattern> patterns = env.socketTextStream("127.0.0.1", 9001).map(new MapFunction<String, Pattern>() {@Overridepublic Pattern map(String value) throws Exception {String[] dataArray = value.split(",");Pattern pattern = new Pattern();pattern.setFirstAction(dataArray[0].trim());pattern.setSecondAction(dataArray[1].trim());return pattern;}});KeyedStream<Action, Long> actionByUser = actions.keyBy(action -> action.userId);MapStateDescriptor<Void,Pattern> bcStateDescriptor = new MapStateDescriptor<Void, Pattern>("patterns", Types.VOID,Types.POJO(Pattern.class));//广播模式pattern状态BroadcastStream<Pattern> bcedPatterns = patterns.broadcast(bcStateDescriptor);SingleOutputStreamOperator<Tuple2<Long, Pattern>> matches = actionByUser.connect(bcedPatterns).process(new PatternEvaluator());matches.map(new MapFunction<Tuple2<Long, Pattern>, Tuple3<Long,String,String>>() {@Overridepublic Tuple3<Long, String, String> map(Tuple2<Long, Pattern> value) throws Exception {return new Tuple3<>(value.f0,value.f1.getFirstAction(),value.f1.getSecondAction());}}).print();try {env.execute("broadcastJob");} catch (Exception e) {e.printStackTrace();}}
}
class PatternEvaluator extends KeyedBroadcastProcessFunction<Long,Action,Pattern, Tuple2<Long,Pattern>>{//每个用户维护一个上次操作的状态ValueState<String> prevActionState;//广播状态DescriptorMapStateDescriptor<Void,Pattern> patternDesc;@Overridepublic void open(Configuration parameters) throws Exception {prevActionState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastAction",Types.STRING));patternDesc = new MapStateDescriptor<Void, Pattern>("patterns",Types.VOID,Types.POJO(Pattern.class));}//这个方法是针对非广播流的元素到来调用方法，在这里是对于用户的Action@Overridepublic void processElement(Action value, ReadOnlyContext ctx, Collector<Tuple2<Long, Pattern>> out) throws Exception {//获取当前广播过来的模式状态Pattern pattern = ctx.getBroadcastState(this.patternDesc).get(null);//获取前一个用户动作，也就是存在prevActionState的，是上一次用户操作的动作String preAction = prevActionState.value();if(pattern!=null && preAction != null){//如果上一次的动作和模式的第一个动作匹配，而且这一次的动作和模式的第二个动作匹配，那么就是满足情况的if(pattern.firstAction.equals(preAction) && pattern.secondAction.equals(value.action)){out.collect(new Tuple2<>(ctx.getCurrentKey(), pattern));}}//将本次动作更新到状态，作为下一个动作的上一个动作prevActionState.update(value.action);}//这个是每个广播流过来的时候@Overridepublic void processBroadcastElement(Pattern value, Context ctx, Collector<Tuple2<Long, Pattern>> out) throws Exception {BroadcastState<Void, Pattern> bcstate = ctx.getBroadcastState(patternDesc);bcstate.put(null,value);}
}class Action{Long userId;String action;public Action(Long userId, String action) {this.userId = userId;this.action = action;}
}

运行结果：

先来个(a,b)的Pattern：

然后输入几个Action(UserId，action)

得出结果：

如果新来一个Pattern(b，e)

同时新来几个Action（UserId，action）：

结果：

KeyedBroadcastProcessFunction接口

上面是实现了KeyedBroadcastProcessFunction接口
总共有三个方法可以实现：

• processBroadcastElement()方法：这个方法是广播流的数据到来的时候调用的方法。在上面场景中，我们使用的是用MapState来保存广播的状态，用了一个null键，从而可以使只保存一个Pattern
• processElement()方法：这个是每次一个非广播状态的数据到来时可以调用的
• onTimer()方法：这个仍然可以注册定时器，我们前面的场景其实可以优化，就是如果一个用户长时间没用进行操作，就可以清空上次操作的状态。

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！