package com.ctgu.flink.project;
import com.ctgu.flink.entity.BlackInfo;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
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.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
public class Flink_Sql_AdClick {
public static void main(String[] args) throws Exception {
long start = System.currentTimeMillis();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.inStreamingMode()
.build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
String createSql =
"CREATE TABLE source " +
" (" +
" `userId` BIGINT," +
" `adId` BIGINT," +
" `province` STRING," +
" `city` STRING," +
" `ts` BIGINT" +
" )" +
" WITH (" +
" 'connector'='filesystem'," +
" 'format'='csv'," +
" 'csv.field-delimiter'=','," +
" 'path'='data/AdClickLog.csv'" +
" )";
tableEnv.executeSql(createSql);
String targetSql = "select *, ts * 1000 as `timestamp` from source";
Table targetTable = tableEnv.sqlQuery(targetSql);
DataStream<Row> rowDataStream = tableEnv.toDataStream(targetTable);
Table source =
tableEnv.fromDataStream(
rowDataStream,
Schema.newBuilder()
.columnByExpression("time_ltz", "TO_TIMESTAMP_LTZ(`timestamp`, 3)")
.watermark("time_ltz", "time_ltz - INTERVAL '5' SECOND")
.build());
DataStream<Row> dataStream = tableEnv.toDataStream(source);
SingleOutputStreamOperator<Row> filterDataStream = dataStream
.filter(row -> row.getField("userId") != null && row.getField("adId") != null)
.keyBy(new KeySelector<Row, Tuple2<Long, Long>>() {
@Override
public Tuple2<Long, Long> getKey(Row row) throws Exception {
return new Tuple2<>((Long) row.getField("userId"), (Long) row.getField("adId"));
}
}).process(new MyProcessFunction(2L));
DataStream<BlackInfo> blackList =
filterDataStream.getSideOutput(new OutputTag<BlackInfo>("blackList"){});
blackList.print();
filterDataStream
.keyBy(new KeySelector<Row, String>() {
@Override
public String getKey(Row row) throws Exception {
return (String) row.getField("province");
}
})
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.aggregate(new AverageAggregate(), new MyWindowFunction()).print();
env.execute("Table SQL");
System.out.println("耗时: " + (System.currentTimeMillis() - start) / 1000);
}
private static class AverageAggregate
implements AggregateFunction<Row, Long, Long> {
@Override
public Long createAccumulator() {
return 0L;
}
@Override
public Long add(Row row, Long aLong) {
return aLong + 1;
}
@Override
public Long getResult(Long aLong) {
return aLong;
}
@Override
public Long merge(Long a, Long b) {
return a + b;
}
}
private static class MyWindowFunction
implements WindowFunction<Long, Tuple3<String, Long, Long>, String, TimeWindow> {
@Override
public void apply(String key,
TimeWindow timeWindow,
Iterable<Long> iterable,
Collector<Tuple3<String, Long, Long>> collector) throws Exception {
collector.collect(new Tuple3<>(key, timeWindow.getEnd(), iterable.iterator().next()));
}
}
private static class MyProcessFunction
extends KeyedProcessFunction<Tuple2<Long, Long>, Row, Row> {
private Long threshold;
private ValueState<Long> clickCount;
private ValueState<Boolean> isBlack;
public MyProcessFunction(Long threshold) {
this.threshold = threshold;
}
@Override
public void open(Configuration parameters) throws Exception {
clickCount = getRuntimeContext().getState(
new ValueStateDescriptor<>("count", Long.class, 0L));
isBlack = getRuntimeContext().getState(
new ValueStateDescriptor<>("isBlack", Boolean.class, false));
}
@Override
public void processElement(Row row, Context context, Collector<Row> collector) throws Exception {
Long curCount = clickCount.value();
if (curCount == 0) {
Long ts = (context.timerService().currentProcessingTime() / (24 * 60 * 60 * 1000) + 1) * (24 * 60 * 60 * 1000);
context.timerService().registerProcessingTimeTimer(ts);
}
if (!isBlack.value()) {
if (curCount >= threshold) {
isBlack.update(true);
context.output(new OutputTag<BlackInfo>("blackList"){},
new BlackInfo((Long) row.getField("userId"),
(Long) row.getField("adId"),
(String) row.getField("province")));
} else {
curCount += 1;
clickCount.update(curCount);
collector.collect(row);
}
}
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Row> out) throws Exception {
isBlack.clear();
clickCount.clear();
}
}
}
网友评论