Flink SQL 入门：SQL Client、批流执行模式与九类连接器参数详解- 惊觉

一、Flink SQL 是什么：动态表 + 连续查询

Flink SQL 的核心抽象只有两个：

动态表（Dynamic Table）：把无界的数据流看成一张「一直在追加/更新」的表；
连续查询（Continuous Query）：一条 SQL 不是执行一次就结束，而是随着输入数据不断产生结果变更。

因此同一条 SELECT clazz, COUNT(1) FROM students GROUP BY clazz，在批模式下只产出一份最终结果，在流模式下会先输出 (文科一班, 1)，再输出 (文科一班, 2)、(文科一班, 3)，以此类推——每一次变更都是一条 changelog 记录。

二、进入 SQL Client

2.1 两种启动方式

# 方式一：本地环境（嵌入式 MiniCluster）
sql-client.sh

# 方式二：提交到 YARN 会话集群（推荐）
yarn-session.sh -d
sql-client.sh -s yarn-session

-s yarn-session 中的 yarn-session 告诉 SQL Client “以 YARN Session 模式运行”。它默认会去读取本地 /tmp/.yarn-properties-root 文件中记录的最近一次 YARN Session 的 Application ID，从而完成连接。

2.2 结果展示模式

-- 表格形式，适合看静态结果（批模式）
SET 'sql-client.execution.result-mode' = 'table';

-- 类画面板形式，会持续刷新，适合看流式结果
SET 'sql-client.execution.result-mode' = 'tableau';

-- 变更日志形式，直接展示 +I / -U / +U / -D 语义
SET 'sql-client.execution.result-mode' = 'changelog';

模式	适合场景	特点
`table`	批查询、结果集较小	一次性打印成表格，流查询会一直等到作业结束
`tableau`	流查询、想看实时变化	会周期刷新，可直接观察 count 的增长
`changelog`	调试 upsert 语义	显式打印 `+I/-U/+U/-D`，排查主键问题最直观

2.3 常用客户端命令

SHOW CATALOGS;              -- 查看已注册的 catalog
SHOW DATABASES;
SHOW TABLES;
DESCRIBE students;          -- 查看表结构
EXPLAIN SELECT * FROM students;   -- 查看执行计划，调优必用

三、批模式与流模式

-- 流处理模式（实时计算）
-- 1、输出连续的结果，以插入、更新、删除的形式输出
-- 2、每来一条数据都会触发一次计算，可能输出一个结果
-- 3、主要用于处理无界流，也能处理有界流
SET 'execution.runtime-mode' = 'streaming';

-- 批处理模式（离线计算，类似 MR / Spark）
-- 1、只输出最终结果
-- 2、底层按有界数据集做 shuffle 和聚合
-- 3、只能处理有界流
SET 'execution.runtime-mode' = 'batch';

维度	streaming	batch
输出	持续输出变更（+I/-U/+U/-D）	只输出一次最终结果
算子	增量聚合、状态存储	排序、分区、一次性聚合
数据边界	有界、无界均可	必须有界
典型延迟	毫秒~秒级	分钟级起
结果表要求	sink 需支持 upsert 或 changelog 格式	普通追加写即可
典型用途	实时大屏、实时风控	离线报表、T+1 统计

四、表的三种类型

类型	语法	生命周期	用途
永久表（Catalog Table）	`CREATE TABLE t (...)`	存入 catalog（Hive/Paimon），跨会话可见	生产表
临时表（Temporary Table）	`CREATE TEMPORARY TABLE t (...)`	仅当前会话，不落 catalog	中间结果、测试源
视图（View）	`CREATE TEMPORARY VIEW v AS SELECT ...`	仅当前会话，存的是 SQL 文本	复用逻辑

-- 视图：只存 SQL 逻辑，不落数据
CREATE TEMPORARY VIEW v_students AS
SELECT id, name, clazz FROM students WHERE age > 20;

SELECT * FROM v_students;

临时表最大的价值是「不污染元数据」：验证一个 connector 时用 CREATE TEMPORARY TABLE，退出客户端就自动消失。

五、九类连接器拆解

5.1 datagen：造数据的源头

CREATE TABLE students_datagen (
  id STRING,
  name STRING,
  age INT,
  sex STRING,
  clazz STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '5',        -- 每秒生成 5 条
  'fields.id.length' = '10',      -- 字符串随机长度
  'fields.name.length' = '2',
  'fields.sex.length' = '1',
  'fields.clazz.length' = '4',
  'fields.age.min' = '1',         -- 数值随机区间
  'fields.age.max' = '100'
);

SELECT * FROM students_datagen;

datagen 永远是无界流（除非加 LIMIT）。它生成的字符串是随机字符，不是中文名，所以做「按姓名分组」这类测试时结果会很散，这是正常现象。

在时间 $T$ 秒内生成的数据量近似为：

$\approx \text{rows-per-second} \times T$

把 rows-per-second 调到 50000 就是压测的常用手段，配合 BlackHole 可以测纯吞吐。

5.2 filesystem：最通用的文件连接器

有界读（批语义）：

CREATE TABLE students_batch (
  id STRING, name STRING, age INT, sex STRING, clazz STRING
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs://master:9000/data/students',
  'format' = 'csv',
  'csv.field-delimiter' = ','
);

SELECT * FROM students_batch;

无界读（持续监控新文件）：

CREATE TABLE students_stream (
  id STRING, name STRING, age INT, sex STRING, clazz STRING
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs://master:9000/data/students',
  'format' = 'csv',
  'csv.field-delimiter' = ',',
  'source.monitor-interval' = '5000'   -- 每 5 秒扫描一次目录下的新文件
);

SELECT * FROM students_stream;

-- 另开终端上传新文件，查询结果会增量出现
-- hdfs dfs -put a.csv /data/students

写文件：

CREATE TABLE clazz_num_hdfs (
  clazz STRING,
  num BIGINT
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs://master:9000/data/clazz_num',
  'format' = 'json'
);

-- 批模式：追加写 / 覆盖写
SET 'execution.runtime-mode' = 'batch';
INSERT INTO clazz_num_hdfs SELECT clazz, COUNT(1) AS num FROM students_batch GROUP BY clazz;
INSERT OVERWRITE clazz_num_hdfs SELECT clazz, COUNT(1) AS num FROM students_batch GROUP BY clazz;

流模式写更新流，普通 json 格式只能表达「插入」，聚合结果里的 -U/+U 会丢语义。两种正确做法：

-- 做法一：用 canal-json 承载变更语义（每条记录带 type 字段）
-- {"data":[{"clazz":"理科四班","num":91}],"type":"DELETE"}
-- {"data":[{"clazz":"理科四班","num":92}],"type":"INSERT"}
CREATE TABLE clazz_num_hdfs_canal (
  clazz STRING,
  num BIGINT
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs://master:9000/data/clazz_num_hdfs_canal',
  'format' = 'canal-json'
);

-- 做法二：把查询改成只产生 INSERT 流（过滤掉聚合，或只写明细的 where 结果）
CREATE TABLE students_yiban (
  id STRING, name STRING, age INT, sex STRING, clazz STRING
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs://master:9000/data/students_yiban',
  'format' = 'json'
);

INSERT INTO students_yiban SELECT * FROM students_batch WHERE clazz = '文科一班';

文件格式选择：

format	是否支持变更流	典型用途
`csv`	否（仅 append）	简单导入导出
`json`	否（仅 append）	通用数据交换
`canal-json`	是（带 type 字段）	承接 CDC / upsert 结果
`debezium-json`	是	对接 Debezium
`avro` / `parquet` / `orc`	否，但列式压缩好	数仓分层落盘

5.3 JDBC（MySQL）：source 与 sink

-- source：有界流（一次性全表扫描）
CREATE TABLE clazz_num_jdbc (
  clazz STRING,
  num BIGINT,
  PRIMARY KEY (clazz) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://master:3306/bigdata',
  'table-name' = 'clazz_num',
  'username' = 'root',
  'password' = '123456'
);

SELECT * FROM clazz_num_jdbc;

作为 sink 时有一个硬规则：要写更新流，MySQL 表必须有主键，且 Flink 侧声明 PRIMARY KEY ... NOT ENFORCED，Flink 才会生成 INSERT ... ON DUPLICATE KEY UPDATE：

CREATE TABLE students_jdbc (
  id STRING, name STRING, age INT, sex STRING, clazz STRING
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://master:3306/bigdata',
  'table-name' = 'students',
  'username' = 'root',
  'password' = '123456'
);

-- 仅插入的流，直接追加
INSERT INTO students_jdbc SELECT * FROM students_batch;

-- 更新流：主键表 + upsert
CREATE TABLE sex_num (
  sex STRING,
  num BIGINT,
  PRIMARY KEY (sex) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://master:3306/bigdata',
  'table-name' = 'sex_num',
  'username' = 'root',
  'password' = '123456'
);

INSERT INTO sex_num SELECT sex, COUNT(1) AS num FROM students_batch GROUP BY sex;

JDBC 还有两个生产环境必调的参数：'sink.buffer-flush.max-rows'（默认 100）与 'sink.buffer-flush.interval'（默认 1s）。批量越大吞吐越高、延迟越大，按业务取舍。

5.4 Kafka：最常用的源与汇

source：

CREATE TABLE cars (
  car STRING,
  city_code STRING,
  county_code STRING,
  card BIGINT,
  camera_id STRING,
  orientation STRING,
  road_id BIGINT,
  `time` BIGINT,
  speed DOUBLE
) WITH (
  'connector' = 'kafka',
  'topic' = 'cars',
  'properties.bootstrap.servers' = 'master:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json',
  'json.ignore-parse-errors' = 'true'      -- 脏数据跳过而不是让作业挂掉
);

scan.startup.mode 的五个可选值：

值	语义
`group-offsets`	从该消费组已提交的 offset 继续（默认）
`earliest-offset`	从最早可用 offset 开始（等价「从头消费」）
`latest-offset`	从最新 offset 开始（只消费新数据）
`timestamp`	从指定时间戳开始，需配 `scan.startup.timestamp-millis`
`specific-offsets`	从每个分区指定的 offset 开始，需配 `scan.startup.specific-offsets`

按时间戳消费：

CREATE TABLE cars_ts (
  car STRING, city_code STRING, card BIGINT, speed DOUBLE,
  `event_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'   -- 数据进入 Kafka 的时间
) WITH (
  'connector' = 'kafka',
  'topic' = 'cars',
  'properties.bootstrap.servers' = 'master:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1794033642684',
  'format' = 'json',
  'json.ignore-parse-errors' = 'true'
);

SELECT car, event_time FROM cars_ts;

注意： METADATA FROM 'timestamp' 拿的是 Kafka 记录的 timestamp（生产者写入时间或 broker 追加时间，取决于 message.timestamp.type），不是业务字段，两者不要混用。

sink：

CREATE TABLE students_kafka_sink (
  id STRING, name STRING, age INT, sex STRING, clazz STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'students_json',
  'properties.bootstrap.servers' = 'master:9092',
  'properties.group.id' = 'testGroup',   -- sink 端其实用不到，但很多模板会带上
  'format' = 'json'
);

INSERT INTO students_kafka_sink SELECT * FROM students_batch;

-- 另开终端验证
-- kafka-console-consumer.sh --bootstrap-server master:9092 --from-beginning --topic students_json

consumer 的 properties.group.id 对 sink 无意义（sink 是 producer），但保留不会报错。真正影响写 Kafka 的是 sink.partitioner（default / fixed / round-robin）和 properties.acks。

5.5 print：最快的调试出口

CREATE TABLE print_table (
  id STRING, name STRING, age INT, sex STRING, clazz STRING
) WITH ('connector' = 'print');

INSERT INTO print_table SELECT * FROM students_datagen;

print 把每条记录打到 TaskManager 的 stdout（log/flink-*-taskexecutor-*.out），不是客户端。所以在 YARN 上跑时要去 NodeManager 节点看日志；本地 sql-client.sh 跑则直接显示在终端。它还有个 'print-identifier' 参数可以给输出打前缀，多路输出时很有用。

5.6 BlackHole：纯压测

CREATE TABLE blackhole_table (
  id STRING, name STRING, age INT, sex STRING, clazz STRING
) WITH ('connector' = 'blackhole');

INSERT INTO blackhole_table SELECT * FROM students_datagen;

BlackHole 什么都不写，专门用来测「上游 + 计算」的吞吐上限。和 datagen 的 50000 行/秒组合，是判断瓶颈在上游还是下游的标准手法。

5.7 Doris：OLAP 侧落库

先理解定位差异，这决定了建表方式：

	MySQL（OLTP）	Doris（OLAP）
场景	事务、点查、强一致	分析、聚合、大宽表扫描
存储	行存	列存 + 按列压缩
加速手段	B+ 树索引	前缀索引、分区、分桶、向量化执行、物化视图

-- sink 表
CREATE TABLE students_doris (
  id STRING, name STRING, age INT, sex STRING, clazz STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'doris',
  'fenodes' = 'master:8030',
  'table.identifier' = 'bigdata.students',
  'username' = 'root',
  'password' = '123456',
  'sink.label-prefix' = 'doris_label'
);

INSERT INTO students_doris SELECT * FROM students_batch;

Doris 侧建表（Unique Key 模型才支持按主键覆盖）：

CREATE TABLE IF NOT EXISTS students
(
  id varchar(20),
  name STRING,
  age INT,
  sex STRING,
  clazz STRING
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 10
PROPERTIES ("replication_num" = "1");

Doris source 就是把同一张表当 source 读（SELECT * FROM students_doris），底层走 FE 的查询接口。

版本冲突是这一节最常见的事故：connector jar 与 Flink 版本、Doris 版本不匹配时会报 NoSuchMethodError 或 ClassNotFoundException。处理流程：

# 1、确认当前 jar 版本
ls /usr/local/soft/flink-1.18.1/lib | grep doris

# 2、换成匹配当前 Flink 1.18 的 connector 包后，删掉旧包（不要两个版本共存）
# 3、重启会话集群
yarn application -list
yarn application -kill application_1793874687606_0002
yarn-session.sh -d
# 4、sql-client 也要退出重进，否则还是旧的 classloader

关键经验：同一个 connector 只能留一个 jar。两个版本共存时，谁先被加载是随机的，报错信息完全不可预测。

5.8 MySQL CDC：真·增量

CREATE TABLE clazz_num_cdc (
  clazz STRING,
  num BIGINT,
  PRIMARY KEY (clazz) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',    -- 只能读，不能写
  'hostname' = 'master',
  'port' = '3306',
  'username' = 'root',
  'password' = '123456',
  'database-name' = 'bigdata',
  'table-name' = 'clazz_num'
);

SELECT * FROM clazz_num_cdc;

CDC 与 JDBC 的本质差别：

	JDBC source	mysql-cdc
数据边界	有界（快照一次）	无界（快照 + binlog 持续订阅）
能否感知更新	不能	能，输出 +I/-U/+U/-D
依赖	JDBC 驱动	binlog（`ROW` 格式）+ 复制权限
资源占用	每次查询扫表	长连接 + binlog 位点

CDC 上线前要检查三件事：binlog_format=ROW、binlog_row_image=FULL、账号有 SELECT, REPLICATION SLAVE, REPLICATION CLIENT 权限。

5.9 Paimon：湖仓表

CREATE CATALOG paimon_catalog WITH (
  'type' = 'paimon',
  'warehouse' = 'hdfs://master:9000/data/paimon'
);
USE CATALOG paimon_catalog;

CREATE TABLE students_paimon (
  id STRING, name STRING, age INT, sex STRING, clazz STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH ('connector' = 'paimon');

Paimon 写数据依赖 checkpoint；同时它支持「批读 + 流读」两种模式，流读可以拿到快照后的增量变更（scan.mode / scan.snapshot-id 控制）。这也是它能承担「ODS→DWD 实时分层」的原因。

5.10 连接器一览表

connector	读	写	数据边界	支持 upsert
`datagen`	✅	❌	无界	—
`filesystem`	✅	✅	有界（默认）/ 无界（配 monitor-interval）	仅 canal/debezium 格式
`jdbc`	✅	✅	有界	✅（需主键）
`kafka`	✅	✅	无界	✅（upsert-kafka）
`print`	❌	✅	—	仅打印
`blackhole`	❌	✅	—	丢弃
`doris`	✅	✅	有界读 / 无界写	✅（Unique Key 模型）
`mysql-cdc`	✅	❌	无界	✅（输出 changelog）
`paimon`	✅	✅	均可	✅（主键表）

六、并行度

SET 'parallelism.default' = '2';

CREATE TABLE clazz_num_print (
  clazz STRING, num BIGINT
) WITH ('connector' = 'print');

INSERT INTO clazz_num_print
SELECT clazz, COUNT(1) AS num FROM students_datagen GROUP BY clazz;

并行度决定每个算子切几个 subtask，每个 subtask 占一个 slot。并行度大于 slot 数时作业无法启动（一直等资源）；并行度过小时单核成瓶颈。经验值：源端并行度对齐 Kafka 分区数，sink 端对齐目标表的分桶数（Doris BUCKETS、Paimon bucket）。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/qq_74148481/article/details/170560417

Flink SQL 入门：SQL Client、批流执行模式与九类连接器参数详解