导航
Flink SQL Client 使用参考
最近更新时间：2024.01.24 15:06:29首次发布时间：2022.11.30 10:22:00
Flink 官方提供的 SQL 客户端可以支持编写 SQL、调试和提交 Flink 任务到 Flink 集群上的功能，具体使用操作，可参考Flink官方文档。
本文将额外介绍几种火山引擎 E-MapReduce（EMR）Flink 的使用场景。
1 Flink SQL Client 运行模式
在火山 EMR Flink 下，我们可以通 SQL 客户端将 Flink SQL 任务提交到 standlone 集群或者 Yarn 集群。
1.1 Standlone 集群场景
Standlone 场景下需要先启动一个 Standlone 的集群，可在FLINK_HOME目录(默认为/usr/lib/emr/current/flink)下运行以下命令创建：
./bin/start-cluster.sh
Standlone 集群启动成功后，可以执行以下命令启动 SQL 客户端命令行界面：
./bin/sql-client.sh embedded
如果想停止 Standlone 集群，可执行以下命令停止：
./bin/stop-cluster.sh
1.2 Yarn 集群场景
Yarn 集群场景下支持多种 Flink 任务提交模式，包括 Yarn-Session 模式，Per-Job Cluster 模式，Application 模式。
Flink SQL Client 暂不支持 Application模式
1.2.1 Session 模式
Session 模式下，需要先执行以下命令启动 Yarn Session：
./bin/yarn-session.sh -d
Yarn Session 启动成功后，会创建一个/tmp/.yarn-properties-root文件，记录最近一次提交到 Yarn 的 Application ID，执行以下命令启动 SQL 客户端命令行界面，后续指定的 Flink SQL 会提交到之前启动的 Yarn Session Application。
./bin/sql-client.sh embedded -s yarn-session
可以执行以下命令停止当前启动的 Yarn Session
cat /tmp/.yarn-properties-root | grep applicationID | cut -d'=' -f 2 | xargs -I {} yarn application -kill {}
1.2.2 Per-Job Cluster 模式
Per-Job Cluster 模式无需提前启动集群，可以在启动 SQL 客户端命令行界面，设置execution.target，后续提交的每一个 Flink SQL 任务将会作为独立的任务提交到 Yarn。
说明
yarn-per-job 模式已经在 Flink 1.16 被标记为 deprecated 状态。
./bin/sql-client.sh embedded Flink SQL> set execution.target=yarn-per-job; [INFO] Session property has been set.
也可以通过在flink-conf.yaml文件预定义配置改参数
# flink-conf.yaml execution.target: yarn-per-job
2 Flink SQL 集成 TOS
火山 EMR Flink 支持多种方式对 TOS 对象存储进行读写操作，比如基于Hive Connector，Hudi Connector等。
2.1 Flink SQL 集成 Hive Connector
以下以 yarn-session 模式为例，显示如何集成 Hive Connector。
- 启动 SQL 客户端命令行界面
./bin/yarn-session.sh -d ./bin/sql-client.sh embedded -s yarn-session -j connectors/flink-sql-connector-hive-3.1.2_2.12-1.16.1.jar
- 创建 Hive Catalog 以及 Demo 数据库
Flink SQL> CREATE CATALOG hive WITH ( > 'type' = 'hive', > 'hive-conf-dir' = '/etc/emr/hive/conf' > ); [INFO] Execute statement succeed. Flink SQL> use catalog hive; [INFO] Execute statement succeed. Flink SQL> create database demo_db; [INFO] Execute statement succeed. Flink SQL> show databases; +---------------+ | database name | +---------------+ | default | | demo_db | +---------------+ 3 rows in set
- 建表
可以通过其它引擎创建 Hive 表，比如 Spark、Hive 等，也可以在 Flink SQL 客户端切换到 Hive Dialect 模式。
- Spark 引擎建表
# 启动Spark SQL命令行交互界面 spark-sql spark-sql> CREATE TABLE demo_tbl1 ( > uuid STRING, > name STRING, > age INT, > ts TIMESTAMP > ) > PARTITIONED BY (`partition` STRING); Time taken: 0.652 seconds spark-sql> desc formatted demo_tbl1; uuid string name string age int ts timestamp partition string # Partition Information # col_name data_type comment partition string # Detailed Table Information Database demo_db Table demo_tbl1 Created Time Fri Nov 25 15:04:43 CST 2022 Last Access UNKNOWN Created By Spark 3.2.1 Type MANAGED Provider hive Comment Table Properties [bucketing_version=2, sink.partition-commit.policy.kind=metastore, transient_lastDdlTime=1669359883] Location tos://xxxxx-v2/hms-warehouse/demo_db.db/demo_tbl1 Serde Library org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe InputFormat org.apache.hadoop.mapred.TextInputFormat OutputFormat org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat Storage Properties [serialization.format=1] Partition Provider Catalog Time taken: 0.085 seconds, Fetched 25 row(s)
- Flink Hive Dialect 建表
从 Flink 1.16 开始，如果要使用 Hive Dialect 模式，需要对 Flink 系统部分 Jar 依赖进行调整，详情参考 官方文档，可执行以下命令进行依赖 Jar 包准备。
# 进入FLINK_HOME mv lib/flink-table-planner-loader-1.16.1.jar opt/ cp opt/flink-table-planner_2.12-1.16.1.jar lib/ cp connectors/flink-sql-connector-hive-3.1.2_2.12-1.16.1.jar lib/
依赖 Jar 包调整之后，需重启 yarn-session 以及 sql-client，建表内容如下：
./bin/sql-client.sh embedded -s yarn-session Flink SQL> set table.sql-dialect=hive; [INFO] Session property has been set. Flink SQL> CREATE CATALOG hive WITH ( > 'type' = 'hive', > 'hive-conf-dir' = '/etc/emr/hive/conf' > ); [INFO] Execute statement succeed. Flink SQL> use catalog hive; [INFO] Execute statement succeed. CREATE TABLE `hive`.`demo_db`.`demo_tbl2` ( uuid STRING, name STRING, age INT, ts TIMESTAMP ) PARTITIONED BY (`partition` STRING) LOCATION 'tos://{bucket_name}/hms-warehouse/demo_db.db/demo_tbl2' TBLPROPERTIES ( 'sink.partition-commit.policy.kind'='metastore' );
建议非 hive dialect 场景，不要将 connector 的 Jar 包依赖放入 lib 目录下。
- 读写数据
- 批式读写数据
Flink SQL> SET 'execution.runtime-mode' = 'batch'; # 建议配置flink-conf.yaml中 [INFO] Session property has been set. Flink SQL> INSERT INTO `hive`.`demo_db`.`demo_tbl1` VALUES > ('id21','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'), > ('id22','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'), > ('id23','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'), > ('id24','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'), > ('id25','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'), > ('id26','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'), > ('id27','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'), > ('id28','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4'); [INFO] Submitting SQL update statement to the cluster... Flink SQL> SET 'sql-client.execution.result-mode' = 'tableau'; # 建议配置在flink-conf.yaml中 [INFO] Session property has been set. Flink SQL> select * from `hive`.`demo_db`.`demo_tbl1`; +------+---------+-----+-------------------------------+-----------+ | uuid | name | age | ts | partition | +------+---------+-----+-------------------------------+-----------+ | id27 | Bob | 44 | 1970-01-01 00:00:07.000000000 | par4 | | id26 | Emma | 20 | 1970-01-01 00:00:06.000000000 | par3 | | id22 | Stephen | 33 | 1970-01-01 00:00:02.000000000 | par1 | | id25 | Sophia | 18 | 1970-01-01 00:00:05.000000000 | par3 | | id21 | Danny | 23 | 1970-01-01 00:00:01.000000000 | par1 | | id28 | Han | 56 | 1970-01-01 00:00:08.000000000 | par4 | | id23 | Julian | 53 | 1970-01-01 00:00:03.000000000 | par2 | | id24 | Fabian | 31 | 1970-01-01 00:00:04.000000000 | par2 | +------+---------+-----+-------------------------------+-----------+ 8 rows in set
- 流式读写数据
对分区表进行流式写入时，需要设置分区提交策略，通知下游某个分区已经写完毕可以被读取了。非分区表可以不设置，亦可以在建表时设置到表的 properties 中。
# 切换到Streaming模式 Flink SQL> SET 'execution.runtime-mode' = 'streaming'; [INFO] Session property has been set. # 对分区表，修改表Properties，或者在建表时设置该属性 Flink SQL> ALTER TABLE `hive`.`demo_db`.`demo_tbl1` SET ('sink.partition-commit.policy.kind'='metastore'); [INFO] Execute statement succeed. Flink SQL> INSERT INTO `hive`.`demo_db`.`demo_tbl1` VALUES > ('id11','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'), > ('id12','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'), > ('id13','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'), > ('id14','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'), > ('id15','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'), > ('id16','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'), > ('id17','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'), > ('id18','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4'); [INFO] Submitting SQL update statement to the cluster... Flink SQL> select * from `hive`.`demo_db`.`demo_tbl1`; +----+--------------------------------+--------------------------------+-------------+-------------------------------+--------------------------------+ | op | uuid | name | age | ts | partition | +----+--------------------------------+--------------------------------+-------------+-------------------------------+--------------------------------+ | +I | id14 | Fabian | 31 | 1970-01-01 00:00:04.000000000 | par2 | | +I | id28 | Han | 56 | 1970-01-01 00:00:08.000000000 | par4 | | +I | id26 | Emma | 20 | 1970-01-01 00:00:06.000000000 | par3 | | +I | id13 | Julian | 53 | 1970-01-01 00:00:03.000000000 | par2 | | +I | id12 | Stephen | 33 | 1970-01-01 00:00:02.000000000 | par1 | | +I | id15 | Sophia | 18 | 1970-01-01 00:00:05.000000000 | par3 | | +I | id18 | Han | 56 | 1970-01-01 00:00:08.000000000 | par4 | | +I | id11 | Danny | 23 | 1970-01-01 00:00:01.000000000 | par1 | | +I | id27 | Bob | 44 | 1970-01-01 00:00:07.000000000 | par4 | | +I | id22 | Stephen | 33 | 1970-01-01 00:00:02.000000000 | par1 | | +I | id17 | Bob | 44 | 1970-01-01 00:00:07.000000000 | par4 | | +I | id25 | Sophia | 18 | 1970-01-01 00:00:05.000000000 | par3 | | +I | id24 | Fabian | 31 | 1970-01-01 00:00:04.000000000 | par2 | | +I | id16 | Emma | 20 | 1970-01-01 00:00:06.000000000 | par3 | | +I | id23 | Julian | 53 | 1970-01-01 00:00:03.000000000 | par2 | | +I | id21 | Danny | 23 | 1970-01-01 00:00:01.000000000 | par1 | +----+--------------------------------+--------------------------------+-------------+-------------------------------+--------------------------------+ Received a total of 16 rows
2.2 Flink SQL 集成 Hudi Connector
Flink 直接集成 Hudi 进行读写 TOS 操作，并将元数据同步到 HMS 中，供其它引擎查询。下面以 Yarn per-job 为例，演示相关集成操作。
Flink 集成 Hudi 需要引入 hudi-flink-bundle 包，目前在 EMR 集群启用 Hudi 的场景下，默认已经提供 hudi-flink-bundle 包。
- 启动 SQL 客户端命令行页面
./bin/sql-client.sh embedded -j connectors/flink-sql-connector-hive-3.1.2_2.12-1.16.1.jar
- 创建 Hudi 表，开启 HMS Hive Sync 同步
以创建 COW 表为例，并持久化 Flink 创建的 Hudi 表在 HMS 中。Flink 集成 Hudi 详细细节可参考Hudi官方文档，Flink 集成 Hudi 开启 Hive Sync 详细细节可参考 Hudi官方文档。
CREATE CATALOG hive WITH ( 'type' = 'hive', 'hive-conf-dir' = '/etc/emr/hive/conf' ); CREATE TABLE `hive`.`demo_db`.`flink_hudi_cow_tbl_hms_sync_demo_original` ( uuid VARCHAR(20), name VARCHAR(10), age INT, ts TIMESTAMP(3), `partition` VARCHAR(20) ) PARTITIONED BY (`partition`) WITH ( 'connector' = 'hudi', 'path' = 'tos://{bucket_name}/hms-warehouse/demo_db.db/flink_hudi_cow_tbl_hms_sync_demo', 'table.type' = 'COPY_ON_WRITE', 'hive_sync.enable' = 'true', 'hive_sync.mode' = 'hms', 'hive_sync.metastore.uris' = 'thrift://{ip}:{port}', 'hive_sync.table'='flink_hudi_cow_tbl_hms_sync_demo', 'hive_sync.db'='demo_db' );
- 使用 Flink SQL 进行读写操作
向 COW 表中写入数据，数据写入成功后，会将元数据同步到hive_sync.table指定的 table 中，COW 场景下只会创建一个目标表，MOR 场景下除了目标表以外，还会额外创建RO和RT表。
INSERT INTO `hive`.`demo_db`.`flink_hudi_cow_tbl_hms_sync_demo_original` VALUES ('id41','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'), ('id42','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'), ('id43','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'), ('id44','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'), ('id45','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'), ('id46','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'), ('id47','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'), ('id48','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4'); Flink SQL> show tables; +-------------------------------------------+ | table name | +-------------------------------------------+ | flink_hudi_cow_tbl_hms_sync_demo | | flink_hudi_cow_tbl_hms_sync_demo_original | +-------------------------------------------+ Flink SQL> select * from `hive`.`demo_db`.`flink_hudi_cow_tbl_hms_sync_demo_original`; +----+--------------------------------+--------------------------------+-------------+-------------------------+--------------------------------+ | op | uuid | name | age | ts | partition | +----+--------------------------------+--------------------------------+-------------+-------------------------+--------------------------------+ | +I | id38 | Han | 56 | 1970-01-01 00:00:08.000 | par4 | | +I | id37 | Bob | 44 | 1970-01-01 00:00:07.000 | par4 | | +I | id31 | Danny | 23 | 1970-01-01 00:00:01.000 | par1 | | +I | id32 | Stephen | 33 | 1970-01-01 00:00:02.000 | par1 | | +I | id35 | Sophia | 18 | 1970-01-01 00:00:05.000 | par3 | | +I | id36 | Emma | 20 | 1970-01-01 00:00:06.000 | par3 | | +I | id33 | Julian | 53 | 1970-01-01 00:00:03.000 | par2 | | +I | id34 | Fabian | 31 | 1970-01-01 00:00:04.000 | par2 | +----+--------------------------------+--------------------------------+-------------+-------------------------+--------------------------------+
- 使用 Spark 引擎进行读操作
spark-sql --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \ --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \ --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' spark-sql> use demo_db; spark-sql> show tables; flink_hudi_cow_tbl_hms_sync_demo flink_hudi_cow_tbl_hms_sync_demo_original Time taken: 0.072 seconds, Fetched 2 row(s) spark-sql> select * from flink_hudi_cow_tbl_hms_sync_demo; 20221125181029843 20221125181029843_2_0 id31 par1 1bde4613-4260-4349-bf4a-23a4a5d35c77_2-4-0_20221126092842836.parquet id31 Danny 231970-01-01 08:00:01 par1 20221125181029843 20221125181029843_2_1 id32 par1 1bde4613-4260-4349-bf4a-23a4a5d35c77_2-4-0_20221126092842836.parquet id32 Stephen 331970-01-01 08:00:02 par1 20221126092842836 20221126092842836_2_2 id42 par1 1bde4613-4260-4349-bf4a-23a4a5d35c77_2-4-0_20221126092842836.parquet id42 Stephen 331970-01-01 08:00:02 par1 20221126092842836 20221126092842836_2_3 id41 par1 1bde4613-4260-4349-bf4a-23a4a5d35c77_2-4-0_20221126092842836.parquet id41 Danny 231970-01-01 08:00:01 par1 20221125181029843 20221125181029843_1_0 id33 par2 4bd429d0-6af4-47c5-829f-887568c015f7_1-4-0_20221126092842836.parquet id33 Julian 531970-01-01 08:00:03 par2 20221125181029843 20221125181029843_1_1 id34 par2 4bd429d0-6af4-47c5-829f-887568c015f7_1-4-0_20221126092842836.parquet id34 Fabian 311970-01-01 08:00:04 par2 20221126092842836 20221126092842836_1_2 id44 par2 4bd429d0-6af4-47c5-829f-887568c015f7_1-4-0_20221126092842836.parquet id44 Fabian 311970-01-01 08:00:04 par2 20221126092842836 20221126092842836_1_3 id43 par2 4bd429d0-6af4-47c5-829f-887568c015f7_1-4-0_20221126092842836.parquet id43 Julian 531970-01-01 08:00:03 par2 20221125181029843 20221125181029843_0_0 id35 par3 8f4f152c-898c-44f1-aeea-4d8e158a24ff_0-4-0_20221126092842836.parquet id35 Sophia 181970-01-01 08:00:05 par3 20221125181029843 20221125181029843_0_1 id36 par3 8f4f152c-898c-44f1-aeea-4d8e158a24ff_0-4-0_20221126092842836.parquet id36 Emma 201970-01-01 08:00:06 par3 20221126092842836 20221126092842836_0_2 id46 par3 8f4f152c-898c-44f1-aeea-4d8e158a24ff_0-4-0_20221126092842836.parquet id46 Emma 201970-01-01 08:00:06 par3 20221126092842836 20221126092842836_0_3 id45 par3 8f4f152c-898c-44f1-aeea-4d8e158a24ff_0-4-0_20221126092842836.parquet id45 Sophia 181970-01-01 08:00:05 par3 20221125181029843 20221125181029843_3_0 id38 par4 24b3d004-5111-4ecd-8bb3-aa8ee80d5f7a_3-4-0_20221126092842836.parquet id38 Han 561970-01-01 08:00:08 par4 20221125181029843 20221125181029843_3_1 id37 par4 24b3d004-5111-4ecd-8bb3-aa8ee80d5f7a_3-4-0_20221126092842836.parquet id37 Bob 441970-01-01 08:00:07 par4 20221126092842836 20221126092842836_3_2 id48 par4 24b3d004-5111-4ecd-8bb3-aa8ee80d5f7a_3-4-0_20221126092842836.parquet id48 Han 561970-01-01 08:00:08 par4 20221126092842836 20221126092842836_3_3 id47 par4 24b3d004-5111-4ecd-8bb3-aa8ee80d5f7a_3-4-0_20221126092842836.parquet id47 Bob 441970-01-01 08:00:07 par4 Time taken: 7.876 seconds, Fetched 16 row(s)
2.3 Flink SQL 集成 Doris Connector
火上EMR Flink支持通过Flink Doris Connector将数据写入Doris中，并在connector目录下内置了Doris Connector。下面以Yarn per-job为例，演示通过Flink读写Doris。
- 启动SQL客户端命令行页面
# 注意不同EMR版本，connector包的版本有差异，可根据实际connector的版本进行调整 /usr/lib/emr/current/flink/bin/sql-client.sh embedded \ -j connectors/flink-doris-connector-1.16-1.3.0-ve-1.jar
- 远程连接Doris，并创建目标表。
可参考火山EMR-Doris-基础使用，创建对应用户账号。
-- mysql -hxxx.xxx.xx -P9030 -uxxxx -p CREATE DATABASE demo_db; USE demo_db; CREATE TABLE all_employees_info ( emp_no int NOT NULL, birth_date date, first_name varchar(20), last_name varchar(20), gender char(2), hire_date date ) UNIQUE KEY(`emp_no`, `birth_date`) DISTRIBUTED BY HASH(`birth_date`) BUCKETS 1 PROPERTIES ( "replication_allocation" = "tag.location.default: 1" );
- Flink建表Doris Sink表，并插入数据
-- 创建Doris Sink表 CREATE TABLE doris_sink ( emp_no int , birth_date STRING, first_name STRING, last_name STRING, gender STRING, hire_date STRING ) WITH ( 'connector' = 'doris', 'fenodes' = '192.168.1.41:8030', 'table.identifier' = 'demo_db.all_employees_info', 'username' = 'test_user', 'password' = 'test_passwd', 'sink.properties.two_phase_commit'='true', 'sink.label-prefix'='doris_demo_emp_001' ); Flink SQL> INSERT INTO `doris_sink` VALUES (10001,'1953-09-02','Georgi','Facello','M','1986-06-26'), (10002,'1964-06-02','Bezalel','Simmel','F','1985-11-21'), (10003,'1959-12-03','Parto','Bamford','M','1986-08-28'), (10004,'1954-05-01','Chirstian','Koblick','M','1986-12-01'), (10005,'1955-01-21','Kyoichi','Maliniak','M','1989-09-12'), (10006,'1953-04-20','Anneke','Preusig','F','1989-06-02'), (10007,'1957-05-23','Tzvetan','Zielinski','F','1989-02-10'), (10008,'1958-02-19','Saniya','Kalloufi','M','1994-09-15'), (10009,'1952-04-19','Sumant','Peac','F','1985-02-18'), (10010,'1963-06-01','Duangkaew','Piveteau','F','1989-08-24');
- Flink创建Doris Source表，并读取数据
-- 创建Doris Source表 Flink SQL> CREATE TABLE doris_source ( emp_no int , birth_date DATE, first_name STRING, last_name STRING, gender STRING, hire_date DATE ) WITH ( 'connector' = 'doris', 'fenodes' = '192.168.1.41:8030', 'table.identifier' = 'demo_db.all_employees_info', 'username' = 'test_user', 'password' = 'test_passwd', 'doris.read.field' = 'emp_no,birth_date,first_name,last_name,gender,hire_date' ); -- 读取数据 Flink SQL> select * from doris_source; +----+-------------+------------+--------------------------------+--------------------------------+--------------------------------+------------+ | op | emp_no | birth_date | first_name | last_name | gender | hire_date | +----+-------------+------------+--------------------------------+--------------------------------+--------------------------------+------------+ | +I | 10001 | 1953-09-02 | Georgi | Facello | M | 1986-06-26 | | +I | 10002 | 1964-06-02 | Bezalel | Simmel | F | 1985-11-21 | | +I | 10003 | 1959-12-03 | Parto | Bamford | M | 1986-08-28 | | +I | 10004 | 1954-05-01 | Chirstian | Koblick | M | 1986-12-01 | | +I | 10005 | 1955-01-21 | Kyoichi | Maliniak | M | 1989-09-12 | | +I | 10006 | 1953-04-20 | Anneke | Preusig | F | 1989-06-02 | | +I | 10007 | 1957-05-23 | Tzvetan | Zielinski | F | 1989-02-10 | | +I | 10008 | 1958-02-19 | Saniya | Kalloufi | M | 1994-09-15 | | +I | 10009 | 1952-04-19 | Sumant | Peac | F | 1985-02-18 | | +I | 10010 | 1963-06-01 | Duangkaew | Piveteau | F | 1989-08-24 | +----+-------------+------------+--------------------------------+--------------------------------+--------------------------------+------------+ Received a total of 10 rows
2.4 Flink SQL集成StarRocks Connector
与Doris Connector类似，火上EMR Flink支持通过Flink StarRocks Connector将数据写入StarRocks中，并在connector目录下内置了StarRocks Connector。下面以Yarn per-job为例，演示通过Flink读写StarRocks。
- 启动SQL客户端命令行页面
# 注意不同EMR版本，connector包的版本有差异，可根据实际connector的版本进行调整 /usr/lib/emr/current/flink/bin/sql-client.sh embedded \ -j connectors/flink-starrocks-connector-1.16-1.2.5-ve-1.jar set execution.target=yarn-per-job;
- 远程连接StarRocks，并创建目标表
可参考火山EMR-StarRocks-基础使用，创建对应用户账号。
-- mysql -hxxx.xxx.xx -P9030 -uxxxx -p mysql> CREATE DATABASE demo_db; mysql> USE demo_db; mysql> CREATE TABLE all_employees_info ( emp_no int NOT NULL, birth_date date, first_name varchar(20), last_name varchar(20), gender char(2), hire_date date ) UNIQUE KEY(`emp_no`, `birth_date`) DISTRIBUTED BY HASH(`birth_date`) BUCKETS 1 PROPERTIES ( "replication_allocation" = "tag.location.default: 1" );
- Flink 创建 StarRocks Sink 表，并插入数据
-- 创建Sink表 Flink SQL> CREATE TABLE sr_sink ( emp_no int , birth_date STRING, first_name STRING, last_name STRING, gender STRING, hire_date STRING ) WITH ( 'connector' = 'starrocks', 'load-url' = '192.168.1.26:8030', 'jdbc-url' = 'jdbc:mysql://192.168.1.26:9030', 'database-name' = 'demo_db', 'table-name' = 'all_employees_info', 'username' = 'test_user', 'password' = 'test_passwd' ); Flink SQL> INSERT INTO `sr_sink` VALUES (10001,'1953-09-02','Georgi','Facello','M','1986-06-26'), (10002,'1964-06-02','Bezalel','Simmel','F','1985-11-21'), (10003,'1959-12-03','Parto','Bamford','M','1986-08-28'), (10004,'1954-05-01','Chirstian','Koblick','M','1986-12-01'), (10005,'1955-01-21','Kyoichi','Maliniak','M','1989-09-12'), (10006,'1953-04-20','Anneke','Preusig','F','1989-06-02'), (10007,'1957-05-23','Tzvetan','Zielinski','F','1989-02-10'), (10008,'1958-02-19','Saniya','Kalloufi','M','1994-09-15'), (10009,'1952-04-19','Sumant','Peac','F','1985-02-18'), (10010,'1963-06-01','Duangkaew','Piveteau','F','1989-08-24');
- Flink创建StarRocks Source表，并查询数据
-- 创建Source表 Flink SQL> CREATE TABLE sr_source ( emp_no int , birth_date STRING, first_name STRING, last_name STRING, gender STRING, hire_date STRING ) WITH ( 'connector' = 'starrocks', 'scan-url' = '192.168.1.26:8030', 'jdbc-url' = 'jdbc:mysql://192.168.1.26:9030', 'database-name' = 'demo_db', 'table-name' = 'all_employees_info', 'username' = 'test_user', 'password' = 'test_passwd' ); Flink SQL> SELECT * FROM sr_source; +----+-------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+ | op | emp_no | birth_date | first_name | last_name | gender | hire_date | +----+-------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+ | +I | 10001 | 1953-09-02 | Georgi | Facello | M | 1986-06-26 | | +I | 10002 | 1964-06-02 | Bezalel | Simmel | F | 1985-11-21 | | +I | 10003 | 1959-12-03 | Parto | Bamford | M | 1986-08-28 | | +I | 10004 | 1954-05-01 | Chirstian | Koblick | M | 1986-12-01 | | +I | 10005 | 1955-01-21 | Kyoichi | Maliniak | M | 1989-09-12 | | +I | 10006 | 1953-04-20 | Anneke | Preusig | F | 1989-06-02 | | +I | 10007 | 1957-05-23 | Tzvetan | Zielinski | F | 1989-02-10 | | +I | 10008 | 1958-02-19 | Saniya | Kalloufi | M | 1994-09-15 | | +I | 10009 | 1952-04-19 | Sumant | Peac | F | 1985-02-18 | | +I | 10010 | 1963-06-01 | Duangkaew | Piveteau | F | 1989-08-24 | +----+-------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+ Received a total of 10 rows
2.5 Flink SQL 集成 Bytehouse(云数仓版)
火上 Bytehouse(云数仓版)支持通过 Flink ByteHouse-Cdw Connector 将数据写入 ByteHouse(云数仓版)中，并在 connectors 目录下内置了 Bytehouse-cdw Connector。下面以 Yarn per-job 为例，演示通过 datagen 写入ByteHouse（云数仓版本）。
参考 ByteHouse云数仓版 / 快速入门 创建bytehouse集群和建表, 其中建库和建表语句如下
create databese flink_bytehouse_test;CREATE TABLE `flink_bytehouse_test`.`cnch_table_test` ( test_key STRING, test_value UInt64, ts UInt64 ) ENGINE = CnchMergeTree ORDER BY (test_key);启动 SQL 客户端命令行页面
# 注意不同EMR版本，connector包的版本有差异，可根据实际connector的版本进行调整 /usr/lib/emr/current/flink/bin/sql-client.sh embedded \ -j connectors/flink-sql-connector-bytehouse-cdw_2.12-1.25.4-1.16.jar set execution.target=yarn-per-job;Flink创建ByteHouse（云数仓版本）Sink表，通过datagen 插入数据到ByteHouse(云数仓表)
CREATE TABLE random_source ( test_key STRING, test_value BIGINT, ts BIGINT ) WITH ( 'connector' = 'datagen', 'rows-per-second' = '1' ); CREATE TABLE cnch_table (test_key STRING, test_value BIGINT, ts BIGINT) WITH ( 'connector' = 'bytehouse-cdw', 'database' = 'flink_bytehouse_test', 'table-name' = 'cnch_table_test', 'bytehouse.gateway.region' = 'VOLCANO', 'bytehouse.gateway.access-key-id' = '<此处填写用户实际的 AK>', 'bytehouse.gateway.secret-key' = '<此处填写用户实际的 SK>' ); INSERT INTO cnch_table SELECT * FROM random_source;
- 进入bytehouse SQL 工作表 页签，单击选择对应角色和正在运行的计算组, 执行查询插入数据的条数
select count(*) from `flink_bytehouse_test`.`cnch_table_test`;
2.6 Flink SQL 集成 Bytehouse(企业版)
类似ByteHouse(云数仓版)， 火上ByteHouse(企业版)支持通过Flink ByteHouse-CE connector将数据写入到ByteHouse(企业版)中。并在 connectors 目录下内置了 Bytehouse-CE Connector(EMR 1.8.0 开始内置)。下面以 Yarn per-job 为例，演示通过 datagen 写入ByteHouse（企业版）。
- 参考ByteHouse（企业版）库表创建，创建bytehouse集群和建表
- 启动 SQL 客户端命令行页面
/usr/lib/emr/current/flink/bin/sql-client.sh embedded \ -j connectors/flink-sql-connector-bytehouse-ce_2.12-1.25.4-1.16.jar set execution.target=yarn-per-job;
- 通过gateway的方式的插入数据到Bytehouse（企业版）
CREATE TABLE random_source ( test_key STRING, test_value BIGINT, ts BIGINT ) WITH ( 'connector' = 'datagen', 'rows-per-second' = '1' ); CREATE TABLE cnch_table (test_key STRING, test_value BIGINT, ts BIGINT) WITH ( 'connector' = 'bytehouse-ce', 'clickhouse.shard-discovery.kind' = 'CE_GATEWAY', 'bytehouse.ce.gateway.host' = '7322262036938936630.bytehouse-ce.ivolces.com', 'bytehouse.ce.gateway.port' = '8123', 'sink.enable-upsert' = 'false', 'clickhouse.cluster' = 'bytehouse_cluster_enct', -- bytehouse集群名称 'database' = 'default', -- 目标数据库 'table-name' = 'cnch_table_test', -- 目标表 注意是local表：{table_name}_local 'username' = 'xxxx@bytedance.com', -- bytehouse 用户 'password' = 'xxx' -- bytehouse 密码 ); INSERT INTO cnch_table SELECT * FROM random_source;
- 进入ByteHouse（企业版） 数据查询页面，选择对应的集群，执行查询插入数据的条数
select count(*) FROM `default`.`cnch_table_test`;