java版用 Flink SQL 搭建一个从Kafka实时统计应用程序【纯干货】

zhengxxm

365人浏览 · 2026-04-08 15:55:11

zhengxxm · 2026-04-08 15:55:11 发布

flink sql建表语句

CREATE TABLE orders (
order_time TIMESTAMP(3),
user_id BIGINT,
amount DOUBLE,
WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND -- 定义水位线处理延迟
) WITH (
'connector' = 'kafka',
'topic' = 'orders',
'properties.bootstrap.servers' = 'master:9092;node1:9092;node2:9092',
'properties.group.id' = 'flink-sql-orders',
'scan.startup.mode' = 'latest-offset',
'format' = 'json',
'json.ignore-parse-errors' = 'true'
);

kafka测试数据：

{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:20"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:01"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:02"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:13"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:24"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:15"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:30:56"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:07"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:28"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:30:59"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:00"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:31"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:12"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:23"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:24"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:35"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:31:26"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:33:55"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:26"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:27"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:28"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:33:59"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:00"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:21"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:22"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:13"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:04"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:25"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:06"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:27"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:08"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:29"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:30"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:41"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:42"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:23"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:24"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:45"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:36"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:37"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:28"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:49"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:50"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:51"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:42"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:33"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:24"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:45"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:46"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:57"}
{"amount":"5","user_id":"100002","order_time":"2026-04-08 15:34:48"}

登录flink sql客户端执行:

[root@master bin]# ./sql-client.sh

▒▓██▓██▒
▓████▒▒█▓▒▓███▓▒
▓███▓░░ ▒▒▒▓██▒ ▒
░██▒ ▒▒▓▓█▓▓▒░ ▒████
██▒ ░▒▓███▒ ▒█▒█▒
░▓█ ███ ▓░▒██
▓█ ▒▒▒▒▒▓██▓░▒░▓▓█
█░ █ ▒▒░ ███▓▓█ ▒█▒▒▒
████░ ▒▓█▓ ██▒▒▒ ▓███▒
░▒█▓▓██ ▓█▒ ▓█▒▓██▓ ░█░
▓░▒▓████▒ ██ ▒█ █▓░▒█▒░▒█▒
███▓░██▓ ▓█ █ █▓ ▒▓█▓▓█▒
░██▓ ░█░ █ █▒ ▒█████▓▒ ██▓░▒
███░ ░ █░ ▓ ░█ █████▒░░ ░█░▓ ▓░
██▓█ ▒▒▓▒ ▓███████▓░ ▒█▒ ▒▓ ▓██▓
▒██▓ ▓█ █▓█ ░▒█████▓▓▒░ ██▒▒ █ ▒ ▓█▒
▓█▓ ▓█ ██▓ ░▓▓▓▓▓▓▓▒ ▒██▓ ░█▒
▓█ █ ▓███▓▒░ ░▓▓▓███▓ ░▒░ ▓█
██▓ ██▒ ░▒▓▓███▓▓▓▓▓██████▓▒ ▓███ █
▓███▒ ███ ░▓▓▒░░ ░▓████▓░ ░▒▓▒ █▓
█▓▒▒▓▓██ ░▒▒░░░▒▒▒▒▓██▓░ █▓
██ ▓░▒█ ▓▓▓▓▒░░ ▒█▓ ▒▓▓██▓ ▓▒ ▒▒▓
▓█▓ ▓▒█ █▓░ ░▒▓▓██▒ ░▓█▒ ▒▒▒░▒▒▓█████▒
██░ ▓█▒█▒ ▒▓▓▒ ▓█ █░ ░░░░ ░█▒
▓█ ▒█▓ ░ █░ ▒█ █▓
█▓ ██ █░ ▓▓ ▒█▓▓▓▒█░
█▓ ░▓██░ ▓▒ ▓█▓▒░░░▒▓█░ ▒█
██ ▓█▓░ ▒ ░▒█▒██▒ ▓▓
▓█▒ ▒█▓▒░ ▒▒ █▒█▓▒▒░░▒██
░██▒ ▒▓▓▒ ▓██▓▒█▒ ░▓▓▓▓▒█▓
░▓██▒ ▓░ ▒█▓█ ░░▒▒▒
▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓ ▓░▒█░

______ _ _ _ _____ ____ _ _____ _ _ _ BETA
| ____| (_) | | / ____|/ __ \| | / ____| (_) | |
| |__ | |_ _ __ | | __ | (___ | | | | | | | | |_ ___ _ __ | |_
| __| | | | '_ \| |/ / \___ \| | | | | | | | | |/ _ \ '_ \| __|
| | | | | | | | < ____) | |__| | |____ | |____| | | __/ | | | |_
|_| |_|_|_| |_|_|\_\ |_____/ \___\_\______| \_____|_|_|\___|_| |_|\__|

Welcome! Enter 'HELP;' to list all available commands. 'QUIT;' to exit.

Flink SQL> CREATE TABLE orders (
> order_time TIMESTAMP(3),
> user_id BIGINT,
> amount DOUBLE,
> WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND -- 定义水位线处理延迟
> ) WITH (
> 'connector' = 'kafka',
> 'topic' = 'orders',
> 'properties.bootstrap.servers' = 'master:9092;node1:9092;node2:9092',
> 'properties.group.id' = 'flink-sql-orders',
> 'scan.startup.mode' = 'latest-offset',
> 'format' = 'json',
> 'json.ignore-parse-errors' = 'true'
> );
[INFO] Execute statement succeed.

执行查询语句

SELECT
TUMBLE_START(order_time, INTERVAL '1' MINUTE) AS window_start,
TUMBLE_END(order_time, INTERVAL '1' MINUTE) AS window_end,
SUM(amount) AS total_sales
FROM orders
GROUP BY TUMBLE(order_time, INTERVAL '1' MINUTE);

查询结果：

window_start window_end total_sales
2026-04-08 15:50:00.000 2026-04-08 15:51:00.000 45.0
2026-04-08 15:51:00.000 2026-04-08 15:52:00.000 290.0
2026-04-08 15:52:00.000 2026-04-08 15:53:00.000 245.0