当前位置：网站首页>hudi记录

hudi记录

2022-06-30 03:03:00 【胖胖胖胖胖虎】

数据结构

在这里插入图片描述

---->>> Flink on hudi

https://www.jianshu.com/p/f509429c2f20

hudi 同步 hive

https://blog.csdn.net/hjl18309163914/article/details/107844269

建表语句 demo

CREATE TABLE sink_order_mysql_goods_order(
     `goods_order_id` bigint COMMENT '自增主键id'
	, `goods_order_uid` string COMMENT '订单uid'
	, `customer_uid` string COMMENT '客户uid'
	, `customer_name` string COMMENT '客户name'
	, `student_uid` string COMMENT '学生uid'
	, `order_status` bigint COMMENT '订单状态 1:待付款 2:部分付款 3:付款审核 4:已付款 5:已取消'
	, `is_end` bigint COMMENT '订单是否完结 1.未完结 2.已完结'
	, `discount_deduction` bigint COMMENT '优惠总金额（单位：分）'
	, `contract_deduction` bigint COMMENT '老合同抵扣金额（单位：分）'
	, `wallet_deduction` bigint COMMENT '钱包抵扣金额（单位：分）'
	, `original_price` bigint COMMENT '订单原价（单位：分）'
	, `real_price` bigint COMMENT '实付金额（单位：分）'
	, `pay_success_time` timestamp(3) COMMENT '完全支付时间'
	, `tags` string COMMENT '订单标签（1新签 2续费 3扩科 4报名-合新 5转班-合新 6续费-合新 7试听-合新）'
	, `status` bigint COMMENT '是否有效（1.生效 2.失效 3.超时未付款）'
	, `remark` string COMMENT '订单备注'
	, `delete_flag` bigint COMMENT '是否删除（1.否，2.是）'
	, `test_flag` bigint COMMENT '是否测试数据（1.否，2.是）'
	, `create_time` timestamp(3) COMMENT '创建时间'
	, `update_time` timestamp(3) COMMENT '更新时间'
	, `create_by` string COMMENT '创建人uid（唯一标识）'
	, `update_by` string COMMENT '更新人uid（唯一标识）'
	, `belong_school` bigint COMMENT '归属校区'
	, `share_sale_no` string COMMENT '共享销售工号'
	
    ,PRIMARY KEY (goods_order_id) NOT ENFORCED
) COMMENT '订单表'
WITH (
      'connector' = 'hudi'
    , 'path' = 'hdfs://hdfs-namenode-service:9000/hudi-warehouse/goods_order' ---路径会自动创建
    , 'hoodie.datasource.write.recordkey.field' = 'goods_order_id'  -- 主键
    , 'write.precombine.field' = 'update_time'             -- 相同的键值时，取此字段最大值，默认ts字段
    , 'read.streaming.skip_compaction' = 'true'            -- 避免重复消费问题
    , 'write.bucket_assign.tasks' = '2'              --并发写的 bucekt 数
    , 'write.tasks' = '2'
    , 'compaction.tasks' = '1'
    , 'write.operation' = 'upsert'                          --UPSERT（插入更新）\INSERT（插入）\BULK_INSERT（批插入）（upsert性能会低些，不适合埋点上报）
    , 'write.rate.limit' = '20000'                          -- 限制每秒多少条
    , 'table.type' = 'COPY_ON_WRITE'                      
    , 'compaction.async.enabled' = 'true'                  -- 在线压缩
    , 'compaction.trigger.strategy' = 'num_or_time'        -- 按次数压缩
    , 'compaction.delta_commits' = '20'               -- 默认为5
    , 'compaction.delta_seconds' = '60'                   -- 默认为1小时
    , 'hive_sync.enable' = 'true'                          -- 启用hive同步
    , 'hive_sync.mode' = 'hms'                             -- 启用hive hms同步，默认jdbc
    , 'hive_sync.metastore.uris' = 'thrift://hive-metastore-svc:9083'    -- required, metastore的端口
    , 'hive_sync.jdbc_url' = 'jdbc:hive2://hive-service-svc:10000'     -- required, hiveServer地址
    , 'hive_sync.table' = 'order_mysql_goods_order'                            -- required, hive 新建的表名 会自动同步hudi的表结构和数据到hive
    , 'hive_sync.db' = 'cdc_ods'                              -- required, hive 新建的数据库名
    , 'hive_sync.username' = 'root'                        -- required, HMS 用户名
    , 'hive_sync.password' = '123456'                            -- required, HMS 密码
    , 'hive_sync.skip_ro_suffix' = 'true'                  -- 去除ro后缀
);

问题记录

hive 同步 hudi 任务报错

CREATE TABLE flink_cdc_sink_hudi_hive(
uuid varchar(20),
name varchar(10),
age int,
ts timestamp(3),
dt varchar(20)
)
PARTITIONED BY (dt)
with(
'connector'='hudi',
'path'= 'hdfs://hdfs-namenode-service:9000/flink_cdc_sink_hudi_hive', 
'table.type'= 'MERGE_ON_READ',
'hoodie.datasource.write.recordkey.field'= 'uuid', 
'write.precombine.field'= 'ts',
'write.tasks'= '1',
'write.rate.limit'= '2000', 
'compaction.tasks'= '1', 
'compaction.async.enabled'= 'true',
'compaction.trigger.strategy'= 'num_commits',
'compaction.delta_commits'= '1',
'changelog.enabled'= 'true',
'read.streaming.enabled'= 'true',
'read.streaming.check-interval'= '3',
'hive_sync.enable'= 'true',
'hive_sync.mode'= 'hms',
'hive_sync.metastore.uris'= 'thrift://hive-metastore-svc:9083',
'hive_sync.jdbc_url'= 'jdbc:hive2://hive-service-svc:10000',
'hive_sync.table'= 'flink_cdc_sink_hudi_hive',
'hive_sync.db'= 'default',
'hive_sync.username'= 'root',
'hive_sync.password'= '123456',
'hive_sync.support_timestamp'= 'true'
);