案例
通过各个广告的点击数量
| 广告id | 某时刻点击数量 |
|---|---|
| 001 | 5 |
| 002 | 7 |
| 003 | 3 |
| 002 | 6 |
| 003 | 4 |
需求
每 5 秒统计一次，统计 最近 10 秒内，各个广告点击的数量
即：size=10，slide=5
提示：基于时间的 滚动 窗口
kafka实现
代码
package day4;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;
public class Std4SlidingProcessTimeWindow {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 定义kafka相关配置项
Properties pro = new Properties();
pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
// 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("guanggao", new SimpleStringSchema(), pro);
DataStreamSource<String> text = env.addSource(fkc);
// 转换操作
SingleOutputStreamOperator<Tuple2<String, Integer>> soso = text
.flatMap(new FlatMapFunImpl())
.keyBy( value -> value.f0) // 根据tuple的第一个元素（广告id）进行分组
// 每滑动5秒，处理最近10秒的数据，即：滑动窗口，窗口大小：10秒，滑动：5秒
.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
.sum(1); // 对 tuple 的第二个元素（索引是1） 求和
soso.print();
System.out.println(env.getExecutionPlan());
env.execute("myjob");
}
// kafka 发送的数据格式是 json，我们接收的是普通字符串，需要转换成 json 对象
static class FlatMapFunImpl implements FlatMapFunction<String,Tuple2<String,Integer>>{
@Override
public void flatMap(String value, Collector<Tuple2<String,Integer>> out) throws Exception {
// 数据样例：001,3
String[] arr = value.split(",");
int num = Integer.parseInt(arr[1]);
// 封装数据为 Tuple2，第一个元素是 广告id，第二个元素是该广告的点击数量
out.collect(Tuple2.of(arr[0],num));
}
}
}
启动 kafka 环境
启动 hadoop1、hadoop2、hadoop3
启动zookeeper
登录 hadoop1，执行下面命令
/program/bin/zk.sh start
启动kafka
登录 hadoop1，执行下面命令
/program/bin/kafka.sh start
创建主题 guanggao
flink 从该主题 读取 数据：
/program/kafka_2.12-2.4.1/bin/kafka-topics.sh --create --zookeeper hadoop1:2181 --replication-factor 3 --partitions 3 --topic guanggao
注意： zookeeper 的主题名称不要有 . 或 _，详见 链接
启动 guanggao 主题生产者
启动另一个窗口执行下面命令：
/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092 --topic guanggao
在上面的环境中，每间隔 5s 输入：
001,3
002,3
003,3
执行结果如下：
(001,3)
(001,3)
(002,3)
(002,3)
(003,3)
(003,3)
每 5 秒统计一次，统计 最近 10 秒内，各个广告点击的数量
如果不输入，统计过去 10 的数据，所以统计结果会下降到 3
注意： 统计结果中，不会是0，因为 没有数据不会触发
socket实现
通过 nc 发送数据，flink 使用 socket 方式接收数据
代码
package top.malaoshi;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
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.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
public class Window统计广告_滑动 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
DataStream<String> dataStream = env.socketTextStream("hadoop1", 9999);
SingleOutputStreamOperator<Tuple2<String, Integer>> soso=dataStream.map(new MapImpl());
KeyedStream<Tuple2<String, Integer>,String> ks=soso.keyBy(new keyByImpl());//流处理中的分组
SingleOutputStreamOperator<Tuple2<String, Integer>> soso2=ks.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
.sum(1);//按照第二个元素聚合
soso2.print();
//启动并等待程序结束，可传入任务名
env.execute("Window WordCount");
}
/**
* 根据 “,” 拆分，返回元祖，格式： (广告id,点击数量)
* MapFunction<String, Tuple2<String, Integer>> 泛型解释：
* String：传入的是字符串
* Tuple2<String, Integer>，返回类型。格式：(广告id,点击数量)
*/
public static class MapImpl implements MapFunction<String, Tuple2<String, Integer>> {
@Override
public Tuple2<String, Integer> map(String s) throws Exception {
String[] arr=s.split(",");
return Tuple2.of(arr[0],Integer.parseInt(arr[1]));
}
}
/**
根据第一个元素（即：广告）进行分组
KeySelector<Tuple2<String, Integer>,String> 泛型解释：
Tuple2<String, Integer>：传入的是 Tuple类型，也就是上面封装的Tuple，格式：(广告id,点击数量)
String：传出类型，也就是Tuple中的单词，根据该单词进行分组
*/
public static class keyByImpl implements KeySelector<Tuple2<String, Integer>,String>{
/**
*
* @param t：传入的元祖，格式：(广告id,点击数量)
* @return
* @throws Exception
*/
@Override
public String getKey(Tuple2<String, Integer> t) throws Exception {
return t.f0;
}
}
}
测试
启动 nc
需要先启动 nc，因为启动 java代码 后，立刻连接 socket 服务器
先执行下面命令：
nc -lk 9999
启动 java 代码
输入
在 nc 窗口中 每间隔 5s 输入：
001,3
002,3
003,3
在 idea 控制台中，显示结果：
8> (003,3)
1> (001,3)
7> (002,3)
7> (002,6)
8> (003,6)
1> (001,6)
8> (003,3)
1> (001,3)
7> (002,3)
每 5 秒统计一次，统计 最近 10 秒内，各个广告点击的数量
如果不输入，统计过去 10 的数据，所以统计结果会下降到 3
注意： 统计结果中，不会是0，因为 没有数据不会触发