Flink高频面试题，来试试你能过关吗？ - IT评测·应用市场-qidao123.com

// 1.准备环境-env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2.准备数据-source
// DataStreamSource<String> inputDataStream = env.socketTextStream("node1.itcast.cn", 9999);
DataStreamSource<String> inputDataStream = env.readTextFile("D:\\0615\\bigdata-flink\\datas\\wordcount.data");
// 3.处理数据-transformation
// TODO: 流计算词频统计WordCount与处理思路基本一致
SingleOutputStreamOperator<Tuple2<String, Integer>> resultDataStream = inputDataStream
// 分割单词
.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String line, Collector<String> out) throws Exception {
for (String word : line.trim().split("\\s+")) {
out.collect(word);
}
}
})
// 转换二元组
.map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String word) throws Exception {
return new Tuple2<>(word, 1);
}
})
// 分组聚合
.keyBy(0).sum(1);
// 4.输出结果-sink
resultDataStream.print();
// 5.触发执行-execute
env.execute(_02StreamWordCount.class.getSimpleName());

复制代码

//1.准备环境流执行环境和流表
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//2.执行SQL,创建 input_kafka 表
TableResult inputTable = tEnv.executeSql(
"CREATE TABLE input_kafka (\n" +
" `user_id` BIGINT,\n" +
" `page_id` BIGINT,\n" +
" `status` STRING\n" +
") WITH (\n" +
" 'connector' = 'kafka',\n" +
" 'topic' = 'input_kafka',\n" +
" 'properties.bootstrap.servers' = 'node1:9092',\n" +
" 'properties.group.id' = 'testGroup',\n" +
" 'scan.startup.mode' = 'latest-offset',\n" +
" 'format' = 'json'\n" +
")"
);
// 创建 output_kafka
TableResult outputTable = tEnv.executeSql(
"CREATE TABLE output_kafka (\n" +
" `user_id` BIGINT,\n" +
" `page_id` BIGINT,\n" +
" `status` STRING\n" +
") WITH (\n" +
" 'connector' = 'kafka',\n" +
" 'topic' = 'output_kafka',\n" +
" 'properties.bootstrap.servers' = 'node1:9092',\n" +
" 'format' = 'json',\n" +
" 'sink.partitioner' = 'round-robin'\n" +
")"
);
// 根据 status 是否为 success 条件筛选出来值
String sql = "select " +
"user_id," +
"page_id," +
"status " +
"from input_kafka " +
"where status = 'success'";
Table ResultTable = tEnv.sqlQuery(sql);
//3.toRetractStream
DataStream<Tuple2<Boolean, Row>> resultDS = tEnv.toRetractStream(ResultTable, Row.class);
//4.打印输出
resultDS.print();
//5.执行sql 将筛选出来success的数据表插入到 output_kafka
tEnv.executeSql("insert into output_kafka select * from "+ResultTable);
//6.excute
env.execute();

复制代码