实时数据及离线数据上云方案

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

CREATE USER canal IDENTIFIED BY 'canal'; 
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

CREATE DATABASE `canal_demo` DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;
CREATE TABLE canal_demo.`student` (
`id`  int(8) NOT NULL auto_increment comment '主键id',
`name`  varchar(64)    comment '名字',
`age`   int(4)     comment '名字',
`gender`  varchar(8)       comment '名字',
PRIMARY KEY (`id`)
) comment '学生表' ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

canal.instance.dbUsername = canal
canal.instance.dbPassword = canal

create table ods_student_canal_sink
(
canal_info string comment 'canal采集信息 json字段'
) comment 'canal采集信息表'
partitioned by (ds string,hh string,mm string);

cd /usr/local/canal/

vim conf/example/instance.properties

/etc/init.d/mysql restart

vim conf/canal.properties

vim conf/kafka_client_producer_jaas.conf

bin/stop.sh
bin/startup.sh

tail -F -n 100 logs/canal/canal.log

tail -F -n 100 logs/example/example.log

tail -F -n 100 logs/example/meta.log

mysql -hlocalhost -P3306 -ucanal -pcanal -Dcanal_demo
show tables;

insert into canal_demo.student (id,name, age, gender) values(1,'张三', 20, '男');

update canal_demo.student set age = 21 where id = 1;

delete from canal_demo.student where id = 1;

truncate table canal_demo.student;

tail -F -n 100 logs/example/meta.log

select * from ods_student_canal_sink where ds <> '';

{
    "data":[
{
"id":"1",
"name":"张三",
"age":"20",
"gender":"男"
}
],
    "database":"canal_demo",
    "es":1658286498000,
    "id":3,
    "isDdl":false,
    "mysqlType":{
"id":"int(8)",
"name":"varchar(64)",
"age":"int(4)",
"gender":"varchar(8)"
},
    "old":null,
    "pkNames":[
"id"
],
    "sql":"",
    "sqlType":{
"id":4,
"name":12,
"age":4,
"gender":12
},
    "table":"student",
    "ts":1658286498272,
    "type":"INSERT"
}

{
"data":[
{
"id":"1",
"name":"张三",
"age":"21",
"gender":"男"
}
],
"database":"canal_demo",
"es":1658286810000,
"id":4,
"isDdl":false,
"mysqlType":{
"id":"int(8)",
"name":"varchar(64)",
"age":"int(4)",
"gender":"varchar(8)"
},
"old":[
{
"age":"20"
}
],
"pkNames":[
"id"
],
"sql":"",
"sqlType":{
"id":4,
"name":12,
"age":4,
"gender":12
},
"table":"student",
"ts":1658286810709,
"type":"UPDATE"
}

{
    "data":[
{
"id":"1",
"name":"张三",
"age":"21",
"gender":"男"
}
],
    "database":"canal_demo",
    "es":1658286829000,
    "id":5,
    "isDdl":false,
    "mysqlType":{
"id":"int(8)",
"name":"varchar(64)",
"age":"int(4)",
"gender":"varchar(8)"
},
    "old":null,
    "pkNames":[
"id"
],
    "sql":"",
    "sqlType":{
"id":4,
"name":12,
"age":4,
"gender":12
},
    "table":"student",
    "ts":1658286829637,
    "type":"DELETE"
}

{
    "data":null,
    "database":"canal_demo",
    "es":1658286836000,
    "id":6,
    "isDdl":true,
    "mysqlType":null,
    "old":null,
    "pkNames":null,
    "sql":"truncate table canal_demo.student",
    "sqlType":null,
    "table":"student",
    "ts":1658286836381,
    "type":"TRUNCATE"
}

-- 全量
CREATE TABLE IF NOT EXISTS ods_sakila_actor_df(
actor_id BIGINT
,first_name STRING
,last_name STRING
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_address_df(
address_id BIGINT
,address STRING
,address2 STRING
,district STRING
,city_id BIGINT
,postal_code STRING
,phone STRING
,location STRING
,last_update DATETIME 
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_category_df(
category_id BIGINT
,name STRING
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_city_df(
city_id BIGINT
,city STRING
,country_id BIGINT
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_country_df(
country_id BIGINT
,country STRING
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_customer_df(
customer_id BIGINT
,store_id BIGINT
,first_name STRING
,last_name STRING
,email STRING
,address_id BIGINT
,active BIGINT
,create_date DATETIME
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_film_df(
film_id BIGINT
,title STRING
,description STRING
,release_year STRING
,language_id BIGINT
,original_language_id BIGINT
,rental_duration BIGINT
,rental_rate DECIMAL
,length BIGINT
,replacement_cost DECIMAL
,rating STRING
,special_features STRING
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_film_actor_df(
actor_id BIGINT
,film_id BIGINT
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_film_category_df(
film_id BIGINT
,category_id BIGINT
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_film_text_df(
film_id BIGINT
,title STRING
,description STRING
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_inventory_df(
inventory_id BIGINT
,film_id BIGINT
,store_id BIGINT
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_language_df(
language_id BIGINT
,name STRING
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_payment_df(
payment_id BIGINT
,customer_id BIGINT
,staff_id BIGINT
,rental_id BIGINT
,amount DECIMAL
,payment_date DATETIME
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_staff_df(
staff_id BIGINT
,first_name STRING
,last_name STRING
,address_id BIGINT
,picture STRING
,email STRING
,store_id BIGINT
,active BIGINT
,username STRING
,password STRING
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_store_df(
store_id BIGINT
,manager_staff_id BIGINT
,address_id BIGINT
,last_update DATETIME
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
-- 增量
CREATE TABLE IF NOT EXISTS stg_sakila_rental_di(
rental_id BIGINT 
,rental_date DATETIME 
,inventory_id BIGINT 
,customer_id BIGINT 
,return_date DATETIME 
,staff_id BIGINT 
,last_update DATETIME 
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------
CREATE TABLE IF NOT EXISTS ods_sakila_rental_df(
rental_id BIGINT 
,rental_date DATETIME 
,inventory_id BIGINT 
,customer_id BIGINT 
,return_date DATETIME 
,staff_id BIGINT 
,last_update DATETIME 
,etl_date DATETIME
)
PARTITIONED BY (ds STRING) 
lifecycle 60;
---------------------------------------------------------------

select 'ods_sakila_actor_df' as table_name, count(1) as record_num from ods_sakila_actor_df where ds = ${bizdate}
union all
select 'ods_sakila_address_df' as table_name, count(1) as record_num from ods_sakila_address_df where ds = ${bizdate}
union all
select 'ods_sakila_category_df' as table_name, count(1) as record_num from ods_sakila_category_df where ds = ${bizdate}
union all
select 'ods_sakila_city_df' as table_name, count(1) as record_num from ods_sakila_city_df where ds = ${bizdate}
union all
select 'ods_sakila_country_df' as table_name, count(1) as record_num from ods_sakila_country_df where ds = ${bizdate}
union all
select 'ods_sakila_customer_df' as table_name, count(1) as record_num from ods_sakila_customer_df where ds = ${bizdate}
union all
select 'ods_sakila_film_df' as table_name, count(1) as record_num from ods_sakila_film_df where ds = ${bizdate}
union all
select 'ods_sakila_film_actor_df' as table_name, count(1) as record_num from ods_sakila_film_actor_df where ds = ${bizdate}
union all
select 'ods_sakila_film_category_df' as table_name, count(1) as record_num from ods_sakila_film_category_df where ds = ${bizdate}
union all
select 'ods_sakila_film_text_df' as table_name, count(1) as record_num from ods_sakila_film_text_df where ds = ${bizdate}
union all
select 'ods_sakila_inventory_df' as table_name, count(1) as record_num from ods_sakila_inventory_df where ds = ${bizdate}
union all
select 'ods_sakila_language_df' as table_name, count(1) as record_num from ods_sakila_language_df where ds = ${bizdate}
union all
select 'ods_sakila_payment_df' as table_name, count(1) as record_num from ods_sakila_payment_df where ds = ${bizdate}
union all
select 'ods_sakila_staff_df' as table_name, count(1) as record_num from ods_sakila_staff_df where ds = ${bizdate}
union all
select 'ods_sakila_store_df' as table_name, count(1) as record_num from ods_sakila_store_df where ds = ${bizdate}
;

mysql -hlocalhost -P3306 -ucanal -pcanal -Dsakila

update rental set last_update = '2022-07-23 11:11:11' where mod(rental_id,100) = 0;

select 'stg_sakila_rental_di' as table_name,count(1) as record_num from stg_sakila_rental_di where ds = '20220723'
union all 
select 'ods_sakila_rental_df' as table_name,count(1) as record_num from ods_sakila_rental_df where ds = '20220722'

INSERT overwrite table ods_sakila_rental_df partition(ds = ${bizdate})
SELECT a.rental_id as rental_id
,a.rental_date as rental_date
,a.inventory_id as inventory_id
,a.customer_id as customer_id
,a.return_date as return_date
,a.staff_id as staff_id
,a.last_update as last_update
,getdate() as etl_date
FROM ods_sakila_rental_df a --全量表
LEFT ANTI JOIN stg_sakila_rental_di b --增量表
ON a.rental_id = b.rental_id
AND b.ds = '${bizdate}'
WHERE a.ds = '${last_day}'
UNION ALL
SELECT b.rental_id as rental_id
,b.rental_date as rental_date
,b.inventory_id as inventory_id
,b.customer_id as customer_id
,b.return_date as return_date
,b.staff_id as staff_id
,b.last_update as last_update
,getdate() as etl_date
FROM stg_sakila_rental_di b
WHERE b.ds = '${bizdate}'
;

select distinct last_update from ods_sakila_rental_df where ds = ${bizdate};