文章參考:https://www.cnblogs.com/ywjfx/p/14722302.html
-- 創建動態表分區
CREATE TABLE t_unity_auth(
`create_time` DATE NOT NULL COMMENT '創建時間',
`route_start` BIGINT(20) NOT NULL COMMENT '請求發起時間',
`route_end` BIGINT(20) NOT NULL COMMENT '請求結束時間',
`account` VARCHAR(32) NOT NULL COMMENT '賬戶'
)
DUPLICATE KEY(create_time,route_start,route_end, account)
PARTITION BY RANGE(create_time)(
PARTITION p20211101 VALUES LESS THAN ("2021-11-02")
)
DISTRIBUTED BY HASH(create_time,product_code) BUCKETS 32
PROPERTIES(
"dynamic_partition.enable" = "true",
"dynamic_partition.time_unit" = "DAY",
"dynamic_partition.end" = "3",
"dynamic_partition.prefix" = "p",
"dynamic_partition.buckets" = "32",
"replication_num" = "1"
);
-- 增加分區語句
ALTER TABLE t_unity_auth ADD
PARTITIONS START ("2021-11-03") END ("2022-11-03") EVERY (interval 1 day);
-- 查詢分區語句
SHOW PARTITIONS FROM t_unity_auth ;
-- 查詢索引語句
SHOW Index FROM t_unity_auth ;
-- 創建ROUTINE LOAD
CREATE ROUTINE LOAD TOPIC_UNITY_AUTH ON t_unity_auth
COLUMNS TERMINATED BY ",",
COLUMNS(create_time,route_id,route_end,account,content_id,product_code,all_pass,result,request_body)
PROPERTIES
(
"desired_concurrent_number"="3",
"max_batch_interval" = "20",
"max_batch_rows" = "300000",
"max_batch_size" = "209715200",
"strict_mode" = "false",
"format" = "json"
)
FROM KAFKA
(
"kafka_broker_list" = "localhost:9095",
"kafka_topic" = "TOPIC_DEMO"
);
-- 查詢ROUTINE LOAD語句
SHOW ROUTINE LOAD;
-- 停止ROUTINE LOAD語句
STOP ROUTINE LOAD FOR TOPIC_DEMO;
-- 重新啟動ROUTINE LOAD語句
RESUME ROUTINE LOAD FOR TOPIC_DEMO;