Из объявления о выпуске Flink 1.5 мы знаем, что Flink теперь поддерживает «широковещательное состояние», и было описано, что «широковещательное состояние разблокирует реализацию функции« динамических шаблонов »для библиотеки Flink CEP».
Означает ли это, что в настоящее время мы можем использовать «состояние широковещания» для реализации «динамических шаблонов» без Flink CEP? Также я понятия не имею, в чем разница при реализации «динамических шаблонов» для Flink CEP с состоянием широковещания или без него? Я был бы признателен, если бы кто-нибудь мог привести пример с кодом, чтобы объяснить разницу.
=============
Обновление для тестирования широковещательного потока данных оператором broadcast () с ключевыми данными
После тестирования в Flink 1.4.2 я обнаружил, что широковещательный поток данных (старый оператор broadcast ()) может подключаться к ключевому потоку данных, ниже - тестовый код, и мы обнаружили, что все события потока управления транслируются на все экземпляры оператора. Таким образом, похоже, что старая трансляция () может выполнять те же функции, что и новое "состояние широковещания".
public static void ConnectBroadToKeyedStream() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(3);
List<Tuple1<String>>
controlData = new ArrayList<Tuple1<String>>();
controlData.add(new Tuple1<String>("DROP"));
controlData.add(new Tuple1<String>("IGNORE"));
DataStream<Tuple1<String>> control = env.fromCollection(controlData);//.keyBy(0);
List<Tuple1<String>>
dataStreamData = new ArrayList<Tuple1<String>>();
dataStreamData.add(new Tuple1<String>("data"));
dataStreamData.add(new Tuple1<String>("DROP"));
dataStreamData.add(new Tuple1<String>("artisans"));
dataStreamData.add(new Tuple1<String>("IGNORE"));
dataStreamData.add(new Tuple1<String>("IGNORE"));
dataStreamData.add(new Tuple1<String>("IGNORE"));
dataStreamData.add(new Tuple1<String>("IGNORE"));
// DataStream<String> data2 = env.fromElements("data", "DROP", "artisans", "IGNORE");
DataStream<Tuple1<String>> keyedDataStream = env.fromCollection(dataStreamData).keyBy(0);
DataStream<String> result = control
.broadcast()
.connect(keyedDataStream)
.flatMap(new MyCoFlatMap());
result.print();
env.execute();
}
private static final class MyCoFlatMap
implements CoFlatMapFunction<Tuple1<String>, Tuple1<String>, String> {
HashSet blacklist = new HashSet();
@Override
public void flatMap1(Tuple1<String> control_value, Collector<String> out) {
blacklist.add(control_value);
out.collect("listed " + control_value);
}
@Override
public void flatMap2(Tuple1<String> data_value, Collector<String> out) {
if (blacklist.contains(data_value)) {
out.collect("skipped " + data_value);
} else {
out.collect("passed " + data_value);
}
}
}
Ниже представлен результат теста.
1> passed (data)
1> passed (DROP)
3> passed (artisans)
3> passed (IGNORE)
3> passed (IGNORE)
3> passed (IGNORE)
3> passed (IGNORE)
3> listed (DROP)
3> listed (IGNORE)
1> listed (DROP)
1> listed (IGNORE)
2> listed (DROP)
2> listed (IGNORE)
https://data-artisans.com/blog/apache-flink-1-5-0-release-announcement
