Skip to content

F
flink_warehouse_rt
Commit d99030f0 authored by 刘喆's avatar 刘喆
Browse files
Options

update bimapfunction

parent 2863b3ed
Hide whitespace changes
Inline Side-by-side
Showing with 10 additions and 10 deletions
bl_et_pe_preciseexposure_inc_d_distinct/src/main/java/com/gmei/streaming/PreciseExposureStreaming.java
View file @ d99030f0
...@@ -137,24 +137,24 @@ public class PreciseExposureStreaming { ...@@ -137,24 +137,24 @@ public class PreciseExposureStreaming {
.filter(new BlPreciseExposureFilterFunction()).uid("id_blpreciseexposure_filter").setParallelism(parallelism) .filter(new BlPreciseExposureFilterFunction()).uid("id_blpreciseexposure_filter").setParallelism(parallelism)
.keyBy(new BlPreciseExposureKeySelector()); .keyBy(new BlPreciseExposureKeySelector());
SingleOutputStreamOperator<BlPreciseExposureBean> blPreciseExposureStreamBF = blPreciseExposureKeyedStream.process(new BloomFilterFunction()); // SingleOutputStreamOperator<BlPreciseExposureBean> blPreciseExposureStreamBF = blPreciseExposureKeyedStream.process(new BloomFilterFunction());
SingleOutputStreamOperator<BlPreciseExposureBean> blPreciseExposureStreamHLL = blPreciseExposureKeyedStream.process(new HyperLogLogFunction()); // SingleOutputStreamOperator<BlPreciseExposureBean> blPreciseExposureStreamHLL = blPreciseExposureKeyedStream.process(new HyperLogLogFunction());
SingleOutputStreamOperator<BlPreciseExposureBean> blPreciseExposureStreamBM = blPreciseExposureKeyedStream.process(new BitMapFunction()); SingleOutputStreamOperator<BlPreciseExposureBean> blPreciseExposureStreamBM = blPreciseExposureKeyedStream.process(new BitMapFunction());
/* /*
BL层数据下发 BL层数据下发
*/ */
blPreciseExposureStreamBF // blPreciseExposureStreamBF
// .print(); // .print();
.addSink(new BlPreciseExposureMysqlSink(sinkJdbcUrl, sinkBFBlTableName)) // .addSink(new BlPreciseExposureMysqlSink(sinkJdbcUrl, sinkBFBlTableName))
.uid("id_blpreciseexposurebf_sink") // .uid("id_blpreciseexposurebf_sink")
.setParallelism(parallelism); // .setParallelism(parallelism);
blPreciseExposureStreamHLL // blPreciseExposureStreamHLL
// .print(); // .print();
.addSink(new BlPreciseExposureMysqlSink(sinkJdbcUrl, sinkHLLBlTableName)) // .addSink(new BlPreciseExposureMysqlSink(sinkJdbcUrl, sinkHLLBlTableName))
.uid("id_blpreciseexposurehll_sink") // .uid("id_blpreciseexposurehll_sink")
.setParallelism(parallelism); // .setParallelism(parallelism);
blPreciseExposureStreamBM blPreciseExposureStreamBM
// .print(); // .print();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment