@
目录实战
写表
插入和覆盖数据
可以使用INSERT语句向表中插入新行或覆盖表中的现有数据。插入的行可以由值表达式指定,也可以由查询结果指定。语法格式如下,其与标准sql语法一致
INSERT { INTO | OVERWRITE } table_identifier [ part_spec ] [ column_list ] { value_expr | query }
- part_spec:一个可选参数,用于指定分区的键和值对的逗号分隔列表。请注意,可以在分区规范中使用类型化文字(例如,日期' 2023-01-02 ')。语法: PARTITION ( partition_col_name = partition_col_val [ , … ] )
- column_list:一个可选参数,用于指定属于table_identifier表的以逗号分隔的列列表。所有指定的列都应该存在于表中,并且不能相互复制。它包括除静态分区列之外的所有列。列列表的大小应该与VALUES子句或查询中数据的大小完全相同。语法: (col_name1 [, column_name2, …])
- value_expr:指定要插入的值。可以插入显式指定的值或NULL。必须用逗号分隔子句中的每个值。可以指定多个值集来插入多行。目前,Flink不支持直接使用NULL,因此NULL应该通过' cast (NULL AS data_type) '转换为实际数据类型。语法: VALUES ( { value | NULL } [ , … ] ) [ , ( … ) ]
CREATE TABLE demo1 (
user_id BIGINT,
item_id BIGINT,
behavior STRING,
dt STRING,
hh STRING,
PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
);
insert into demo1 values(1,1,'order','2023-08-04','19'),(2,2,'pay','2023-08-04','20');
select * from demo1;
CREATE TABLE demo_p1 (
user_id BIGINT,
item_id BIGINT,
behavior STRING,
dt STRING,
hh STRING,
PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) PARTITIONED BY (dt, hh);
insert into demo_p1 partition(dt='2023-08-04',hh='21') values(3,3,'pv');
insert into demo_p1 select * from demo1;
select * from demo_p1;
覆盖只支持batch模式。覆盖默认情况下,流读取将忽略INSERT OVERWRITE生成的提交。如果想要读取OVERWRITE的提交,可以配置流读覆盖。对于分区表,Paimon的默认覆盖模式是动态分区覆盖(这意味着Paimon只删除出现在覆盖数据中的分区)。可以配置动态分区覆盖来更改它。
RESET 'execution.checkpointing.interval';
SET 'execution.runtime-mode' = 'BATCH';
- 覆盖未分区的表
insert overwrite demo1 values(3,3,'pv','2023-08-04','20');
- 覆盖分区表
insert overwrite demo_p1 select * from demo1;
insert overwrite demo_p1 partition(dt='2023-08-04',hh='20') select user_id,item_id,behavior from demo1;
更新数据
目前,Paimon支持在Flink 1.17及以后的版本中使用UPDATE更新记录。可以在Flink的批处理模式下执行UPDATE。重要的表属性设置,只有主键表支持此特性。
要支持此特性,需要对MergeEngine进行重复数据删除或部分更新。不支持更新主键。语法:UPDATE table_identifier SET column1 = value1, column2 = value2, ... WHERE condition;
update demo_p1 set item_id = 5,behavior='uv' where user_id = 1;
# 比如下面,merge-engine默认就是deduplicate
CREATE TABLE MyTable (
a STRING,
b INT,
c INT,
PRIMARY KEY (a) NOT ENFORCED
) WITH (
'write-mode' = 'change-log',
'merge-engine' = 'deduplicate'
);
- deduplicate:删除重复数据,保留最后一行。
- Partial-update:部分更新非空字段。
- aggregation:聚合具有相同主键的字段。
- first-row:删除重复数据并保留第一行。
详细参数配置可以查看:https://paimon.apache.org/docs/master/maintenance/configurations/
删除数据
Flink1.17+以上SQL支持删除数据。且只有写模式设置为更改日志的表才支持此特性;如果表有主键,则需要对MergeEngine进行重复数据删除以支持此特性。
delete from demo_p1 where behavior = 'pv';
select * from demo_p1;
Merge Into
Paimon通过flink run提交“MERGE - INTO”作业来支持“MERGE INTO”。下载paimon-flink-action文件,无需放到lib目录,与普通开发jar包一样指定路径运行即可
wget https://repository.apache.org/content/groups/snapshots/org/apache/paimon/paimon-flink-action/0.5-SNAPSHOT/paimon-flink-action-0.5-20230804.002229-95.jar
通过Merge Into实现行级别更新,只有主键表支持这个功能,该操作不会产生UPDATE_BEFORE,所以不建议设置' changelog-producer ' = ' input '。合并操作使用“upsert”语义而不是“update”语义,这意味着如果行存在,则执行更新,否则执行插入。例如对于非主键表可以更新每一列,但对于主键表如果希望更新主键,则必须插入具有不同于表中行主键的新行。在这种情况下,“upsert”是有用的。
匹配解释如下:
- 匹配:更改的行来自目标表,并且每个行都可以基于merge-condition和可选的Matched -condition (source∩target)匹配源表的行。
- 不匹配:根据合并条件和可选的不匹配条件(source - target),更改的行来自源表,所有行不能匹配任何目标表行。
- Not-matched-by-source:根据merge-condition和可选的Not-matched-by-source条件(target -source),更改的行来自目标表,并且所有行不能匹配任何源表行。
参数格式:
- Matched-upsert-changes: col =
.col | expression [, …] (Means setting .col with given value. Do not add ‘ .’ before ‘col’.),可以使用' * '来设置具有所有源列的列(要求目标表的模式等于源表的模式)。 - Not-matched-upsert-changes:类似于matched-upsert-changes,但不能引用源表的列或使用' * '。
- insert-values:col1, col2, …, col_end。Must specify values of all columns. For each column, you can reference
.col or use an expression。可以使用' * '插入所有源列(要求目标表的模式等于源表的模式)。 - 不匹配条件不能使用目标表的列来构造条件表达式。
- 不匹配源条件不能使用源表的列来构造条件表达式。
create database test;use test;CREATE TABLE wstest1 ( id INT, ts BIGINT, vc INT, PRIMARY KEY (id) NOT ENFORCED)insert into wstest1 values(1,1,1),(2,2,2),(3,3,3);select * from wstest1;
CREATE TABLE wstest2 ( id INT, ts BIGINT, vc INT, PRIMARY KEY (id) NOT ENFORCED)insert into wstest2 values(2,2,2),(3,3,3),(4,4,4),(5,5,5);select * from wstest2;
./bin/flink run \ ./lib/paimon-flink-action-0.5-20230804.002229-95.jar \ merge-into \ --warehouse hdfs://myns/paimon/hive \ --database test \ --table wstest2 \ --source-table test.wstest1 \ --on "wstest2.id = wstest1.id" \ --merge-actions matched-upsert,matched-delete \ --matched-upsert-condition "wstest2.ts > 2" \ --matched-upsert-set "vc = 100" \ --matched-delete-condition "wstest2.ts <= 2"
select * from wstest2;
查询表
批量查询
Paimon的批处理读取返回表快照中的所有数据。默认情况下,批处理读取返回最新的快照。在sql-client中,设置执行模式为
RESET 'execution.checkpointing.interval';SET 'execution.runtime-mode' = 'batch';
时间旅行
带时间旅行的Paimon批读可以指定一个快照或一个标签,并读取相应的数据。通过查看文件系统存储元数据和数据可以看下上面test库中wstest2表的快照目录可以查看目前有1和2两个快照版本
-- 读取id为1的快照SELECT * FROM wstest2 /*+ OPTIONS('scan.snapshot-id' = '1') */;-- 以Unix毫秒为单位从指定的时间戳读取快照SELECT * FROM wstest2 /*+ OPTIONS('scan.timestamp-millis' = '1691143583490') */;-- 读取标签“my-tag”SELECT * FROM wstest2 /*+ OPTIONS('scan.tag-name' = 'my-tag') */;
读取id和时间戳读取数据如下
批量增量查询
读取开始快照(不包含)和结束快照之间的增量变化。例如:
- “5,10”表示快照5和快照10之间的变化。
- ' TAG1,TAG3 '表示TAG1和TAG3之间的变化。
SELECT * FROM wstest2 /*+ OPTIONS('incremental-between' = '1,2') */;
流式查询
默认情况下,流式读取在第一次启动时生成表上的最新快照,并继续读取最新的更改。
-- Flink SQLSET 'execution.checkpointing.interval'='30s';SET 'execution.runtime-mode' = 'streaming';
可以做流式读取没有快照数据,可以使用最新的扫描模式;连续读取最新更改,而不在开始时生成快照。
SELECT * FROM wstest2 /*+ OPTIONS('scan.mode' = 'latest') */;
如果只想处理今天及以后的数据,可以使用分区过滤器:
SELECT * FROM wstest2 WHERE dt > '2023-08-04'
时间旅行
如果它不是一个分区表,或者不能按分区进行过滤,可以使用Time travel的流读取。
-- 从id为1的快照读取更改SELECT * FROM wstest2 /*+ OPTIONS('scan.snapshot-id' = '1') */;-- 从指定时间戳的快照读取更改SELECT * FROM wstest2 /*+ OPTIONS('scan.timestamp-millis' = '1691143583490') */;-- 在第一次启动时读取快照id 1,并继续读取更改SELECT * FROM wstest2 /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id' = '1') */;
读取数据如下
ConsumerID
这是一个实验性的功能。可以在流式读表时指定消费者id:
SELECT * FROM wstest2 /*+ OPTIONS('consumer-id' = 'myid') */;
当流读取Paimon表时,要记录到文件系统中的下一个快照id。这有几个好处:
- 当前一个作业停止时,新启动的作业可以继续使用前一个进度,而无需从状态恢复。新的读取将从消费者文件中找到的下一个快照id开始读取。
- 在确定快照是否过期时,Paimon查看文件系统中表的所有消费者,如果仍然有消费者依赖于该快照,则该快照将不会在到期时被删除。
- 当没有水印定义时,Paimon表会将快照中的水印传递给下游的Paimon表,这意味着您可以跟踪整个管道的水印进度。
注意:消费者将阻止快照过期,可以指定consumer.expiration-time来管理消费者的生命周期。可以使用给定的消费者ID和下一个快照ID重置消费者。
SELECT * FROM wstest2 /*+ OPTIONS('consumer-id' = 'itxiaoshen') */;insert into wstest2 values(6,6,6),(7,7,7);SELECT * FROM wstest2 /*+ OPTIONS('consumer-id' = 'itxiaoshen') */;
查询优化
强烈建议与查询一起指定分区和主键过滤器,这将加快查询的数据跳过。可以加速数据跳转的过滤函数有:
=
<
<=
>
>=
IN (...)
LIKE 'abc%'
IS NULL
Paimon将按主键对数据进行排序,这加快了点查询和范围查询的速度。当使用复合主键时,查询过滤器最好在主键的最左边形成一个前缀,以获得良好的加速。
假设一个表具有以下表结构:
CREATE TABLE orders ( catalog_id BIGINT, order_id BIGINT, ....., PRIMARY KEY (catalog_id, order_id) NOT ENFORCED -- composite primary key)
通过为主键的最左边的前缀指定一个范围过滤器,查询可以获得很好的加速。
SELECT * FROM orders WHERE catalog_id=1025;SELECT * FROM orders WHERE catalog_id=1025 AND order_id=29495;SELECT * FROM orders WHERE catalog_id=1025 AND order_id>2035 AND order_id<6000;
但是下面的过滤器不能很好地加速查询。
SELECT * FROM orders WHERE order_id=29495;SELECT * FROM orders WHERE catalog_id=1025 OR order_id=29495;
系统表
表指定系统表
表指定的系统表包含每个表的元数据和信息,例如创建的快照和正在使用的选项。用户可以通过批量查询访问系统表。
目前,Flink、Spark和Trino都支持查询系统表。在某些情况下,表名需要用反引号括起来以避免语法解析冲突,例如三重访问模式:
SELECT * FROM my_catalog.my_db.`MyTable$snapshots`;
- 快照表
# 通过快照表可以查询该表的快照历史信息,包括快照中发生的记录计数。select * from wstest2$snapshots;
通过查询快照表,可以了解该表的提交和过期信息以及数据的时间旅行。
- 表模式
可以通过schemas表查询该表的历史模式。
SELECT * FROM wstest2$schemas;
可以连接快照表和模式表以获得给定快照的字段。
SELECT s.snapshot_id, t.schema_id, t.fields FROM wstest2$snapshots s JOIN MyTable$schemas t ON s.schema_id=t.schema_id where s.snapshot_id=1;
- 表选项
可以查询表的选项信息,这些信息是通过选项表从DDL指定的。未显示的选项将是默认值。可以参考[Configuration]。
SELECT * FROM wstest2$options;
- 审计日志表
如果需要审计表的变更日志,可以使用audit_log系统表。通过audit_log表,可以在获取表的增量数据时获取rowkind列。您可以使用该列进行过滤和其他操作,以完成审计。
SELECT * FROM wstest2$audit_log;
> +I:插入操作,新增数据。> -U:使用更新行之前的内容进行更新操作,一条数据的修改会产生两个U 标识符数据。其中-U 含义为修改前数据。> +U:使用更新行的新内容进行更新操作,修改之后的数据。> -D:删除操作,删除的数据。
- 表文件
可以查询指定快照表的文件。
-- 查询最新快照的文件SELECT * FROM wstest2$files;
-- 还可以查询指定快照的文件SELECT * FROM wstest2$files /*+ OPTIONS('scan.snapshot-id'='1') */;
- 表标签
通过标签表可以查询该表的标签历史信息,包括标签基于哪些快照,以及快照的一些历史信息。还可以获得所有标签名称和时间旅行到特定的标签数据名称。
SELECT * FROM wstest2$tags;
- 表消费者
可以查询包含下一个快照的所有消费者。
SELECT * FROM wstest2$consumers;
- 表清单文件
可以查询当前表的最新快照或指定快照中包含的所有清单文件。
-- 查询最新快照的清单信息SELECT * FROM wstest2$manifests;
-- 也可以查询带有指定快照的清单SELECT * FROM wstest2$manifests /*+ OPTIONS('scan.snapshot-id'='1') */;
分区表
可以查询表的分区文件。
SELECT * FROM demo_p1$partitions;
全局系统表
全局系统表包含当前存在的所有表的统计信息;为了方便检索,创建了一个参考系统数据库sys,可以用sql在flink中显示所有全局系统表:
所有选项表,这个表类似于Options table,但是它显示所有的表选项都是all database。
SELECT * FROM sys.all_table_options;
维表
Paimon支持Lookup Join,它用于从Paimon查询数据来补充维度字段,是流查询中的一种连接。连接要求一个表具有处理时间属性,另一个表由查找源连接器提供支持。
在Flink中,Paimon支持对带有主键的表和仅追加表进行查找连接。下面的示例说明了这个特性,创建一个Paimon表并实时更新它。
USE CATALOG fs_catalog;CREATE TABLE customers ( id INT PRIMARY KEY NOT ENFORCED, name STRING, country STRING, zip STRING);-- 启动一个流作业来更新客户表INSERT INTO customers values(1,'zhangsan','china','aaa'),(2,'lisi','china','bbb'),(3,'wangwu','china','ccc');select * from customers;-- 创建一个临时左表,就像从kafkaCREATE TEMPORARY TABLE Orders ( order_id INT, total INT, customer_id INT, proc_time AS PROCTIME()) WITH ( 'connector' = 'datagen', 'rows-per-second'='1', 'fields.order_id.kind'='random', 'fields.order_id.min'='1', 'fields.order_id.max'='1000000', 'fields.total.kind'='sequence', 'fields.total.start'='1', 'fields.total.end'='1000', 'fields.customer_id.kind'='random', 'fields.customer_id.min'='1', 'fields.customer_id.max'='3');select * from Orders;
现在可以在查找连接查询中使用客户。
-- 用客户信息填充每个订单SELECT o.order_id, o.total, c.country, c.zipFROM Orders AS oJOIN customersFOR SYSTEM_TIME AS OF o.proc_time AS cON o.customer_id = c.id;
Lookup Join将在本地维护一个RocksDB缓存,并实时提取表的最新更新。查找连接操作符将只提取必要的数据,因此筛选条件对性能非常重要。此特性仅适用于最多包含数千万条记录的表,以避免过度使用本地磁盘。
如果Orders(主表)join的记录缺失,因为客户(查找表)的相应数据还没有准备好。可以考虑使用Flink的Delayed Retry Strategy For Lookup进行查找。以下选项允许用户微调RocksDB以获得更好的性能,可以在表属性或动态表提示中指定它们。
-- 动态表提示示例SELECT o.order_id, o.total, c.country, c.zipFROM Orders AS o JOIN customers /*+ OPTIONS('lookup.cache-rows'='20000') */FOR SYSTEM_TIME AS OF o.proc_time AS cON o.customer_id = c.id;
CDC集成
通过模式演化,Paimon支持多种方式将数据摄取到Paimon表中;这意味着添加的列将实时同步到Paimon表,并且不会为此重新启动同步作业。目前支持以下同步方式:
- MySQL同步表:将MySQL中的一个或多个表同步到一个Paimon表中。
- MySQL同步数据库:将整个MySQL数据库同步到一个Paimon数据库。
- API同步表:自定义数据流输入到一个Paimon表。
- Kafka同步表:将一个Kafka主题的表同步到一个Paimon表。
- Kafka同步数据库:同步一个包含多个表的Kafka主题或多个包含一个表的主题到一个Paimon数据库。
MySQL
Paimon支持使用更改数据捕获(CDC)同步来自不同数据库的更改,此功能需要Flink及其CDC连接器,准备CDC Bundled Jar。在上一篇我们已经将flink-sql-connector-mysql-cdc-2.4.1.jar拷贝到Flink的Lib目录。
- 同步表:通过在Flink数据流作业中使用mysql-sync-table动作或直接通过Flink运行,用户可以将一个或多个MySQL表同步到一个Paimon表中。
如果指定的Paimon表不存在,则此操作将自动创建表。它的模式将从所有指定的MySQL表派生。如果Paimon表已经存在,它的模式将与所有指定的MySQL表的模式进行比较。
./bin/flink run \ ./lib/paimon-flink-action-0.5-20230804.002229-95.jar \ mysql-sync-table \ --warehouse hdfs://myns/paimon/hive \ --database test \ --table my_users_cdc \ --primary-keys id \ --mysql-conf hostname=192.168.50.95 \ --mysql-conf username=root \ --mysql-conf password=123456 \ --mysql-conf database-name='test' \ --mysql-conf table-name='my_users' \ --catalog-conf metastore=hive \ --catalog-conf uri=thrift://hadoop2:9083 \ --table-conf bucket=4 \ --table-conf changelog-producer=input \ --table-conf sink.parallelism=4
启动后查看下表的信息已经同步
修改MySQL数据库my_users表中id为4的age字段的值从40改为55
可以看到已经获取变更的数据
还可以使用正则表达式设置' database-name '来捕获多个数据库。通过对--mysql-conf database-name='source_db.+' ;一个典型的场景是,一个表' source_table '被分成数据库' source_db1 ', ' source_db2 '…,然后可以同步所有' source_table '的数据到一个Paimon表。
- 同步数据库:通过在Flink数据流作业中使用mysql-sync-database动作或直接通过Flink运行,用户可以将整个MySQL数据库同步到一个Paimon数据库。
只有具有主键的表才会被同步。对于每个要同步的MySQL表,如果对应的Paimon表不存在,该操作将自动创建表。它的模式将从所有指定的MySQL表派生。如果Paimon表已经存在,它的模式将与所有指定的MySQL表的模式进行比较。
./bin/flink run \ ./lib/paimon-flink-action-0.5-20230804.002229-95.jar \ mysql-sync-database \ --warehouse hdfs://myns/paimon/hive \ --database test \ --table-prefix "ods_" \ --table-suffix "_cdc" \ --mysql-conf hostname=192.168.50.95 \ --mysql-conf username=root \ --mysql-conf password=123456 \ --mysql-conf database-name=test \ --catalog-conf metastore=hive \ --catalog-conf uri=thrift://hadoop2:9083 \ --table-conf bucket=4 \ --table-conf changelog-producer=input \ --table-conf sink.parallelism=4
运行后查看已经有对应整库的表了
修改MySQL数据库new_users表中id为1的age字段的值从当前33改为43
可以看到已经获取变更的数据
希望该作业同步包含历史数据的表[order, custom];可以通过从作业的前一个快照中恢复,从而重用作业的现有状态来实现这一点。恢复的作业将首先对新添加的表进行快照,然后继续自动从以前的位置读取变更日志。
./bin/flink run \ --fromSavepoint savepointPath \ ./lib/paimon-flink-action-0.5-20230804.002229-95.jar \ mysql-sync-database \ --warehouse hdfs:///path/to/warehouse \ --database test_db \ --mysql-conf hostname=127.0.0.1 \ --mysql-conf username=root \ --mysql-conf password=123456 \ --mysql-conf database-name=source_db \ --catalog-conf metastore=hive \ --catalog-conf uri=thrift://hive-metastore:9083 \ --table-conf bucket=4 \ --including-tables 'product|user|address|order|custom'
可以设置——mode组合,以启用同步新添加的表而无需重新启动作业。
--including-tables 'tbl.+'
通过将database-name设置为正则表达式,同步作业将捕获匹配数据库下的所有表,并将同名的表合并到一个表中。可以设置——merge-shards false来阻止合并碎片。同步表将被命名为' databaseName_tableName ',以避免潜在的名称冲突。
Kafka
在上一篇我们已经将flink-sql-connector-kafka-1.17.1.jar拷贝到Flink的Lib目录。Flink提供了几种Kafka CDC格式:canal-json, debezium-json,ogg-json,maxwell-json。如果Kafka主题中的消息是使用更改数据捕获(CDC)工具从另一个数据库捕获的更改事件,那么您可以使用Paimon Kafka CDC。将解析后的INSERT、UPDATE、DELETE消息写入paimon表。也即是目前只支持CanalCDC,其他在未来应该会支持
准备
[mysqld]log-bin=mysql-bin # 开启 binlogbinlog-format=ROW # 选择 ROW 模式
授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限
CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
先安装canal
# 先下载canalwget https://github.com/alibaba/canal/releases/download/canal-1.1.6/canal.deployer-1.1.6.tar.gz# 解压tar -xvf canal.deployer-1.1.6.tar.gz
修改canal.properties和instance.properties两个配置文件,vim conf/canal.properties
canal.serverMode = kafkakafka.bootstrap.servers = kafka1:9092,kafka2:9092,kafka3:9092
vim conf/example/instance.properties
canal.instance.master.address=mysqlserver:3306canal.instance.dbUsername=canalcanal.instance.dbPassword=canal# mq configcanal.mq.topic=cc_test2
启动canal
# 启动canal./bin/startup.sh
修改MySQL的account数据库的account_tbl表的数据,先通过
./kafka-console-producer.sh --broker-list kafka1:9092 --topic cc_test2
消费kafka的主题cc_test2数据成功如下,验证canal的配置正确。
{"data":[{"id":"1","user_id":"6","money":"100"}],"database":"account","es":1691400547000,"id":2,"isDdl":false,"mysqlType":{"id":"int","user_id":"varchar(255)","money":"int"},"old":[{"user_id":"5"}],"pkNames":["id"],"sql":"","sqlType":{"id":4,"user_id":12,"money":4},"table":"account_tbl","ts":1691400547176,"type":"UPDATE"}{"data":[{"id":"1","user_id":"7","money":"100"}],"database":"account","es":1691400566000,"id":3,"isDdl":false,"mysqlType":{"id":"int","user_id":"varchar(255)","money":"int"},"old":[{"user_id":"6"}],"pkNames":["id"],"sql":"","sqlType":{"id":4,"user_id":12,"money":4},"table":"account_tbl","ts":1691400566482,"type":"UPDATE"}
- 同步表:通过在Flink数据流作业中使用kafka-sync-table动作或直接通过Flink运行,用户可以将Kafka的一个主题中的一个或多个表同步到一个Paimon表中。
./bin/flink run \ ./lib/paimon-flink-action-0.5-20230804.002229-95.jar \ kafka-sync-table \ --warehouse hdfs://myns/paimon/hive \ --database test \ --table kafka_account_tbl_cdc \ --primary-keys id \ --kafka-conf properties.bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092 \ --kafka-conf topic=cc_test2 \ --kafka-conf properties.group.id=itxs \ --kafka-conf value.format=canal-json \ --catalog-conf metastore=hive \ --catalog-conf uri=thrift://hadoop2:9083 \ --table-conf bucket=4 \ --table-conf changelog-producer=input \ --table-conf sink.parallelism=4
运行后可以到通过Kafka已经将对应表和数据都同步到
如果指定的Paimon表不存在,则此操作将自动创建表。它的模式将从所有指定的Kafka主题的表中派生,它从主题中获得最早的非ddl数据解析模式。如果Paimon表已经存在,它的模式将与所有指定Kafka主题表的模式进行比较。
- 同步库:通过在Flink数据流作业中使用KafkaSyncDatabaseAction或直接通过Flink运行,用户可以将多主题或一个主题同步到一个Paimon数据库。只有具有主键的表才会被同步。
此操作将为所有表构建单个合并接收器。对于每个要同步的Kafka主题的表,如果对应的Paimon表不存在,这个动作将自动创建表,并且它的模式将从所有指定的Kafka主题的表中派生。如果Paimon表已经存在,并且它的模式与从Kafka记录解析的模式不同,这个动作将尝试进行模式进化。
修改canal.mq.topic=cc_test3后重启启动canal
./bin/flink run \ ./lib/paimon-flink-action-0.5-20230804.002229-95.jar \ kafka-sync-database \ --warehouse hdfs://myns/paimon/hive \ --database test \ --table-prefix "ods_" \ --table-suffix "_cdc" \ --kafka-conf properties.bootstrap.servers=192.168.5.120:9092 \ --kafka-conf topic=cc_test3 \ --kafka-conf properties.group.id=itxs \ --kafka-conf scan.startup.mode=earliest-offset \ --kafka-conf value.format=canal-json \ --catalog-conf metastore=hive \ --catalog-conf uri=thrift://hadoop2:9083 \ --table-conf bucket=4 \ --table-conf changelog-producer=input \ --table-conf sink.parallelism=4
运行后,也可以看下后很多ods开发和cdc结尾的表,可以看到
在MySQL的student表修改数据
可以看到在sql-client中查询表已经捕获到最新变更的数据,至此基于Kafka通过MySQL多表已验证完毕。
select * from ods_student_cdc;
支持schema变更
CDC摄取支持有限数量的模式更改,也即是可以自动同步表结构信息。目前,框架不能重命名表,删除列,所以rename table和drop COLUMN的行为将被忽略,rename COLUMN将添加一个新的列。当前支持的模式更改包括:
-
添加列。
-
修改列类型。更具体地说
- 将字符串类型(char, varchar, text)转换为另一种长度更长的字符串类型
- 从二进制类型(binary, varbinary, blob)转换为另一种长度更长的二进制类型
- 从整数类型(tinyint, smallint, int, bigint)转换为另一个范围更大的整数类型
- 从浮点类型(float, double)转换为另一种范围更大的浮点类型;
-
本人博客网站IT小神 www.itxiaoshen.com