性能测试-Oceanus 测试FLink mysql到Iceberg性能

一、任务依赖信息

1、mysql测试库信息

地址：127.0.0.1、gomysql_bdg_test

库：bdg_test

表：order_info1

2、iceberg库

hive地址：thrift://127:7004

catalog-name：spark_catalog

Format版本:v2

库：iceberg_test

表：order_info1

3、kafka信息

地址：腾讯kafka

topic：test_wangshida_order

group：read-test-wang

分区数：10

4、任务信息

全量增量同步：test_mysql_iceberg_order_info3

mysql到kafka：test_mysql_kafka_order

kafka到iceberg：test_kafka_iceberg_order

二、元数据信息

1、mysql建表

CREATE TABLE `order_info1` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `order_number` bigint(20) NOT NULL COMMENT '订单号，由外部提供（全局唯一）',
  `user_id` bigint(20) NOT NULL COMMENT '下单的用户Id student_number'
PRIMARY KEY (`id`),
  UNIQUE KEY `orderInfoNumberUniqueIndex` (`order_number`),
  KEY `idx_create_time` (`create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='订单信息'

2、iceberg建表

CREATE TABLE iceberg_test.order_info1(
  `id` bigint
  ,`order_number` bigint
  ,`user_id` bigint
)USING iceberg
TBLPROPERTIES('format-version'='2','property-version'='2','write.upsert.enabled'='true')

3、kafka信息

CREATE TABLE test_kafka_order(
    `id` bigint
    ,`order_number` bigint
    ,`user_id` bigint
) WITH (
    -- 定义 Kafka 参数
    'connector' = 'kafka',
    'topic' = 'test_wangshida_order',
    'scan.startup.mode' = 'earliest-offset', -- 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets / timestamp 的任何一种
    'properties.bootstrap.servers' = '127:9092',  -- 替换为您的 Kafka 连接地址
    'properties.group.id' = 'read-test-wang', -- 必选参数, 一定要指定 Group ID
    -- 定义数据格式 (JSON 格式)
    'format' = 'json',
    'json.fail-on-missing-field' = 'false',  -- 如果设置为 false, 则遇到缺失字段不会报错。
    'json.ignore-parse-errors' = 'true'    -- 如果设置为 true，则忽略任何解析报错。
);

三、任务配置信息

1、引用jar包

2、高级参数

pipeline.max-parallelism: 2048
security.kerberos.login.principal: bdg_app@EMR
security.kerberos.login.keytab: bdg_app.keytab
security.kerberos.login.conf: krb5.conf
containerized.taskmanager.env.HADOOP_USER_NAME: bdg_app
containerized.master.env.HADOOP_USER_NAME: bdg_app
rest.flamegraph.enabled: true

四、任务运行信息

1、全量同步性能

资源情况如下，全量同步完成后可以缩减资源运行。可以全量阶段加资源，全量完成后保存checkpoint，然后缩减资源重启任务。

2、增量同步性能

3、更新数据性能

五、参数调优

修改参数：ALTER TABLE order_info3 SET TBLPROPERTIES('write.distribution-mode'='hash');

1、写入参数介绍

2、表行为属性

其它属性参考官网地址

六、测试结论

checkpoint设置1分钟场景下，一般情况下一个checkpoint就写入
1、一次写入10条，则下个checkpoint就进入
2、连续1分钟写一次，下个checkpoint就进入
3、删除数据，下个checkpoint就逻辑删除

七、更新数据

1、更新数据
update order_info1 set purchase_info_id='test6' where id>10000000 and id<=10100000;
update order_info1 set purchase_info_id='test6' where id>10100000 and id<=10200000;
update order_info1 set purchase_info_id='test6' where id>10200000 and id<=10300000;
update order_info1 set purchase_info_id='test6' where id>10300000 and id<=10400000;
update order_info1 set purchase_info_id='test6' where id>10400000 and id<=10500000;
update order_info1 set purchase_info_id='test6' where id>10500000 and id<=10600000;
update order_info1 set purchase_info_id='test6' where id>10600000 and id<=10700000;
update order_info1 set purchase_info_id='test6' where id>10700000 and id<=10800000;
update order_info1 set purchase_info_id='test6' where id>10800000 and id<=10900000;
update order_info1 set purchase_info_id='test6' where id>10900000 and id<=11000000;
2、插入数据
insert into order_info1 (order_number,user_id,sub_clazz_number,clazz_number,course_number,purchased_minutes,follow_order,order_status,time_expire,purchase_info_id,paid_time,pay_channel,create_time,original_price,spike_price_id,voucher_price,user_voucher_id,manual_price,price,refund_price,refund_time,order_type,parent_order_number,update_time,user_address_id,pay_number,coupon_price,activity_number,course_original_price,course_price,activity_type,activity_ext_info,activity_price,business_type,replay_limit,biz_line,app_version,app_id,service_period) SELECT id as order_number,user_id,sub_clazz_number,clazz_number,course_number,purchased_minutes,follow_order,order_status,time_expire,purchase_info_id,paid_time,pay_channel,create_time,original_price,spike_price_id,voucher_price,user_voucher_id,manual_price,price,refund_price,refund_time,order_type,parent_order_number,update_time,user_address_id,pay_number,coupon_price,activity_number,course_original_price,course_price,activity_type,activity_ext_info,activity_price,business_type,replay_limit,biz_line,app_version,app_id,service_period from order_info1 where id>10000000 and id<=12000000;

八、参考文档

1、iceberg官网-建表

https://iceberg.apache.org/docs/latest/configuration/

2、iceberg官网-flink写入

https://iceberg.apache.org/docs/latest/flink-connector/

本栏目推荐文章
• Jmeter 如何连接mysql数据库？
• 如何通过Navicat把MySQL远程数据库导入到本地数据库
• SQL Server 高性能写入的一些总结
• 高性能、低功耗 基于8位AVR®RISC的ATMEGA328PB-MUR、ATMEGA328PB-ANR、ATMEGA328PB-AUR微控制器（MCU）
• 【LeetCode 2701. 连续递增交易】MySQL用户变量编程得到严格递增连续子序列的开始位置，结束位置，长度等统计信息
• 【Leetcode 2474. 购买量严格增加的客户】MySQL用户变量编程解决严格递增连续子序列问题
• mysql发生连接异常Cause: com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure
• 【LeetCode 2494. 合并在同一个大厅重叠的活动】MySQL用户变量编程解决区间合并问题
• 如何处理Mysql死锁
• 【LeetCode1747. 应该被禁止的 Leetflex 账户】MySQL用户变量编程；尝试维护一个multiset

性能 Oceanus Iceberg FLink mysql性能oceanus iceberg flink oceanus iceberg工具mysql 性能iceberg方式apache javaapi iceberg impala flink 文件 格式flink mysql oceanus catalog flink mysql 1.17 连接器mysql-cdc参数flink 程序flink mysql kafka starrocks flink mysql cdc