案例
每5秒钟，统计一次，买家买了几本数据
最大延迟时间：5秒
数据文件
数据如下：
java从入门到精通,50,李雷,1
hadoop从入门到精通,80,李雷,2
flink从入门到精通,70,韩梅梅,4
mysql从入门到精通,100,李雷,15
hbase从入门到精通,50,李雷,1
hive从入门到精通,100,韩梅梅,3
html从入门到精通,120,韩梅梅,13
解释：
- 第一列：书名
- 第二列：价格
- 第三列：买家姓名
- 第四列：购买时间，这里为了简单，用的是 当前时间距离
1970-01-01 00:00:00秒的秒数
迟到数据：下面数据迟到了
hbase从入门到精通,50,李雷,1
hive从入门到精通,100,韩梅梅,3
html从入门到精通,120,韩梅梅,13
java代码
package day7;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
public class Std4watermark {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
StreamTableEnvironment stEnv = StreamTableEnvironment.create(env);
/*
创建表
注意：表名不要出现关键字，如：order等，否则会报错
*/
TableResult table=stEnv.executeSql("CREATE TABLE t_order ( \n" +
" name STRING, \n" +
" price double, \n" +
" username STRING, \n" +
" ts BIGINT," + // 现在时间距离1970-01-01 00:00:00 的间隔
" event_ts AS TO_TIMESTAMP( from_unixtime(ts)),"+ // 定义事件时间，根据 ts （数据中的时间戳） 转成 TIMESTAMP
" WATERMARK FOR event_ts AS event_ts - INTERVAL '5' SECOND"+ // 水位线延迟5秒
") WITH ( \n" +
"'connector' = 'kafka', \n"+//表示连接kafka
"'topic' = 'user-order-input', \n"+//主题
"'properties.bootstrap.servers' = 'hadoop1:9092', \n"+//服务器
"'properties.group.id' = 'testGroup', \n"+//组
// "'scan.startup.mode' = 'earliest-offset', "+//从最早位置消费
"'scan.startup.mode' = 'latest-offset', \n"+//从最新位置消费
"'format' = 'csv' \n"+
")");
// 统计各个用户买了几本书
String sql = " SELECT " +
" TUMBLE_START(event_ts, INTERVAL '5' second) as w_start, " + //获取窗口的开始时间
" TUMBLE_ROWTIME(event_ts, INTERVAL '5' second) as w_end, " + //获取窗口的结束时间
" username," +
" COUNT(name) as cnt " +
" FROM t_order" +
" GROUP BY username," + // 根据用户名分组
" TUMBLE(event_ts, INTERVAL '5' second)"; // 指定窗口，间隔5秒
System.out.println("sql:"+sql);
Table res=stEnv.sqlQuery(sql);
System.out.println("Schema信息:");
//打印Schema
res.printSchema();
//将计算后的数据，append到ds
DataStream<Row> dsRes2 = stEnv.toAppendStream(res, Row.class);
dsRes2.print();
env.execute();
}
}
测试类
在 flink1.12.x WaterMark 案例：kafka生产者 工程下创建
package test;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import util.TextFileReader;
import java.util.List;
import java.util.Properties;
public class 订单2 {
final static String TOPIC = "user-order-input";
public static void main(String[] args) throws Exception {
// 1. 创建 kafka 生产者的配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
// key,value 序列化（必须）：key.serializer，value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
// StringSerializer.class.getName() 等于 org.apache.kafka.common.serialization.StringSerializer
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer") ;
// 3. 创建 kafka 生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties);
// 4. 调用 send 方法,发送消息
TextFileReader textFileReader = new TextFileReader("./data/订单2.txt");
List<String> list = textFileReader.readList();
for (String line : list) {
kafkaProducer.send(new ProducerRecord(TOPIC, line));
Thread.sleep(500); // 间隔时间不能少于200ms
}
// 5. 关闭资源
kafkaProducer.close();
}
}
执行结果
1970-01-01T08:00,1970-01-01T08:00:04.999,李雷,2
1970-01-01T08:00,1970-01-01T08:00:04.999,韩梅梅,1
解释：
收到 mysql从入门到精通,100,李雷,15 数据后，watermark 是 15-5=10，大于窗口结束时间，才会触发计算
下面数据迟到丢弃
hbase从入门到精通,50,李雷,1
hive从入门到精通,100,韩梅梅,3
html从入门到精通,120,韩梅梅,13