一、环境版本

环境	版本
Flink	1.17.0
Kafka	2.12
MySQL	5.7.33

【注意】Flink 1.13版本增加Cumulate Window，之前版本Flink Sql 没有 Trigger 功能，长时间的窗口不能在中途触发计算，输出中间结果。比如每 10S 更新一次截止到当前的pv、uv。只能用Trigger配合State实现，可参考如下实现方式：
Flink DataStream 按分钟或日期统计数据量

二、MySQL建表脚本

create table user_log
(id      int auto_increment comment '主键'primary key,uid     int    not null comment '用户id',event   int    not null comment '用户行为',logtime bigint null comment '日志时间'
)comment '用户日志表，作为验证数据源';

三、用户日志类

新建maven项目

用以定义Kafka和MySQL中Schema

/*** 用户日志类*/
@Data
public class UserLog {//用户uidprivate int uid;//用户行为private int event;//日志时间private Date logtime;//获取日期，用于按日期统计数据public String getFormatDate() {return DateUtil.format(logtime, "yyyyMMdd");}//获取时间，精确到分钟public String getFormatTime() {return DateUtil.format(logtime, "yyyy-MM-dd HH:mm") + ":00";}
}
}

四、用户数据生成器

/*** 用户数据生成器*/
public class UserLogGenerator {public static void main(String[] args) throws Exception {// 1.获取执行环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// 2.自定义数据生成器SourceDataGeneratorSource<UserLog> dataGeneratorSource = new DataGeneratorSource<>(// 指定GeneratorFunction 实现类new GeneratorFunction<Long, UserLog>(){// 定义随机数数据生成器public RandomDataGenerator generator;@Overridepublic void open(SourceReaderContext readerContext) throws Exception {generator = new RandomDataGenerator();}@Overridepublic UserLog map(Long aLong) throws Exception {UserLog userLog = new UserLog();//随机生成用户uiduserLog.setUid(generator.nextInt(1, 50));//随机生成用户行为userLog.setEvent(generator.nextInt(1, 2));//随机生成用户数据时间userLog.setLogtime(DateUtil.offset(new DateTime(), DateField.MILLISECOND, generator.nextInt(-2000, 2000)));return userLog;}},// 指定输出数据的总行数
//                60 * 60 * 10,1200,// 指定每秒发射的记录数RateLimiterStrategy.perSecond(10),// 指定返回值类型， 将Java的StockPrice封装成到TypeInformationTypeInformation.of(UserLog.class));DataStreamSource<UserLog> dataGeneratorSourceStream = env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "dataGeneratorSource");//输出生成数据
//        dataGeneratorSourceStream.print();//kafka数据写入KafkaSink<UserLog> kafkaSink = KafkaSink.<UserLog>builder().setBootstrapServers("hadoop01:9092").setRecordSerializer(KafkaRecordSerializationSchema.<UserLog>builder().setTopic("userLog").setValueSerializationSchema((SerializationSchema<UserLog>) userLog -> JSONUtil.toJsonStr(userLog).getBytes()).build()).build();dataGeneratorSourceStream.sinkTo(kafkaSink);//MySQL数据写入，用以数据验证SinkFunction<UserLog> jdbcSink = JdbcSink.sink("insert into user_log (uid, event, logtime) values (?, ?, ?)",new JdbcStatementBuilder<UserLog>() {@Overridepublic void accept(PreparedStatement preparedStatement, UserLog userLog) throws SQLException {preparedStatement.setInt(1, userLog.getUid());preparedStatement.setInt(2, userLog.getEvent());preparedStatement.setLong(3, userLog.getLogtime().getTime());}},JdbcExecutionOptions.builder().withBatchSize(1000).withBatchIntervalMs(200).withMaxRetries(5).build(),new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl("jdbc:mysql://192.168.31.116:3306/demo").withDriverName("com.mysql.cj.jdbc.Driver").withUsername("root").withPassword("root").build());dataGeneratorSourceStream.addSink(jdbcSink);env.execute();}
}

五、Sql按分钟或日期统计PV和UV

public class UserLogSql {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);env.setParallelism(1);// 创建一个输入表SourceTableString sourceDDL = "create table user_log\n" +"(\n" +"    uid  INT\n" +"    , event INT\n" +"    , logtime BIGINT\n" +"    , rowtime AS TO_TIMESTAMP_LTZ(logtime, 3)\n" +"    , WATERMARK FOR rowtime AS rowtime - INTERVAL '5' SECOND\n" +") with (\n" +"      'connector' = 'kafka'\n" +"      ,'topic' = 'userLog'\n" +"      ,'properties.bootstrap.servers' = 'hadoop01:9092'\n" +"      ,'scan.startup.mode' = 'latest-offset'\n" +"      ,'format' = 'json'\n" +");";tableEnv.executeSql(sourceDDL);// 统计每分钟PV和UVString result = "select\n" +" date_format(window_start, 'yyyy-MM-dd') cal_day\n" +" , date_format(window_start, 'HH:mm:ss') start_time\n" +" , date_format(window_end, 'HH:mm:ss') end_time\n" +" , count(uid) pv\n" +" , count(distinct uid) uv\n" +"FROM TABLE(\n" +// 每隔10秒触发一次计算，窗口大小为1天
//                "    CUMULATE(TABLE user_log, DESCRIPTOR(rowtime), INTERVAL '10' SECOND, INTERVAL '1' DAY))\n" +// 每隔10秒触发一次计算，窗口大小为10秒"    CUMULATE(TABLE user_log, DESCRIPTOR(rowtime), INTERVAL '10' SECOND, INTERVAL '10' SECOND))\n" +"  GROUP BY window_start, window_end\n" +";";// 输出sql执行结果tableEnv.executeSql(result).print();}
}

六、sql-client方式执行Sql

# 建表语句
create table user_log
(uid  INT,event INT,logtime BIGINT,rowtime AS TO_TIMESTAMP_LTZ(logtime, 3) ,WATERMARK FOR rowtime AS rowtime - INTERVAL '5' SECOND
) with ('connector' = 'kafka','topic' = 'userLog''properties.bootstrap.servers' = 'hadoop01:9092','scan.startup.mode' = 'latest-offset','format' = 'json',
);# pv、uv计算语句, 每隔10秒触发一次计算，窗口大小为1天
selectdate_format(window_start, 'yyyy-MM-dd') cal_day,date_format(window_start, 'HH:mm:ss') start_time,date_format(window_end, 'HH:mm:ss') end_time,count(uid) pv,count(distinct uid) uv
FROM TABLE(CUMULATE(TABLE user_log, DESCRIPTOR(rowtime), INTERVAL '10' SECOND, INTERVAL '1' DAY))GROUP BY window_start, window_end;

七、数据验证

1. 启动 UserLogGenerator
2. 启动 UserLogSql或在sql-client执行Sql
3. 在MySQL中验证查询

转换时间戳

时间戳	转换前	转换后
w_start	2025-08-16 14:45:40	1755326740000
w_end	2025-08-16 14:45:50	1755326750000

select count(distinct uid) from user_log where logtime< 1755326750000 and logtime>=1755326740000;

# 与MySql中输出一致SQL Query Result (Table)                                                               Refresh: 1 s                                                      Page: Last of 1                                              Updated: 23:50:09.972 cal_day                     start_time                       end_time                   pv                   uv2025-08-15                       23:45:30                       23:45:40                   15                   152025-08-15                       23:45:40                       23:45:50                  101                   452025-08-15                       23:45:50                       23:46:00                  104                   422025-08-15                       23:46:00                       23:46:10                  100                   422025-08-15                       23:46:10                       23:46:20                   97                   452025-08-15                       23:46:20                       23:46:30                  104                   402025-08-15                       23:46:30                       23:46:40                   97                   422025-08-15                       23:46:40                       23:46:50                   99                   442025-08-15                       23:46:50                       23:47:00                  103                   442025-08-15                       23:47:00                       23:47:10                   97                   442025-08-15                       23:47:10                       23:47:20                  100                   43

八、常见问题

1. sql-client执行查询，缺少kafka包

# 运行SQL命令
Flink SQL> select * from user_log;
# 报错
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'kafka' that implements 'org.apache.flink.table.factories.DynamicTableFactory' in the classpath.

解决方法

# 下载flink对应版本的kafka包，放到flink的lib目录下
wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/1.17.0/flink-sql-connector-kafka-1.17.0.jar -P ${FLINK_HOME}/lib/

九、参考鸣谢

Flink 实时统计历史 pv、uv
Flink Cumulate Window