通过waterdrop把HIVE的数据同步到clickhouse

- 日理万妓 2022-12-05 13:57 225阅读 0赞

\-- 第一步，在hive建一个 ck临时表，同时把sql在easyschedular ,命名为 dwd\_ord\_car\_sharing\_df\_ck.hql  
drop table cdm\_dwd.dwd\_ord\_car\_sharing\_df\_ck;  
create table cdm\_dwd.dwd\_ord\_car\_sharing\_df\_ck stored as orc as  
select  
order\_id ,  
p\_order\_id ,  
order\_no ,  
merchant\_order\_no ,  
title ,  
user\_no ,  
order\_category ,  
platform ,  
merchant\_id ,  
merchant\_name ,  
merchant\_img ,  
order\_status ,  
sub\_order\_status ,  
sub\_order\_status\_str ,  
total\_fee ,  
wait\_fee ,  
discount\_fee ,  
real\_fee ,  
pay\_status ,  
pay\_type ,  
pay\_id ,  
pay\_no ,  
pay\_time ,  
is\_invoice ,  
refund\_status ,  
comment\_status ,  
device\_platform ,  
create\_time ,  
update\_time ,  
update\_version ,  
is\_delete ,  
is\_pre ,  
cancel\_type ,  
settle\_fee\_status ,  
amount ,  
discount\_amount ,  
coupon\_discount ,  
coupon\_create\_time ,  
coupon\_promotion\_id ,  
coupon\_promotion\_name ,  
coupon\_type ,  
coupon\_title ,  
car\_id ,  
car\_gear ,  
car\_name ,  
car\_seats ,  
car\_emission ,  
fetch\_address ,  
return\_address ,  
fetch\_time ,  
return\_time ,  
ckbizdate ,  
bizdate  
from cdm\_dwd.dwd\_ord\_car\_sharing\_df  
\-- where ckbizdate='2019-10-22 ;  
where ckbizdate=$\{ckbizdate\} ;

\--- 第二步  
\---通过执行创表语句的udf函数 select default.tableinfo("cdm\_dwd.dwd\_ord\_car\_sharing\_df\_ck");  
获取执行语句.

==============请确定主键和date类型的列,并把Nullable去掉,比如本例是 order\_id=============  
CREATE TABLE cdm\_dwd.dwd\_ord\_car\_sharing\_df on cluster crm\_4shards\_1replicas (  
\`order\_id\` String,  
\`p\_order\_id\` Nullable(String),  
\`order\_no\` Nullable(String),  
\`merchant\_order\_no\` Nullable(String),  
\`title\` Nullable(String),  
\`user\_no\` Nullable(String),  
\`order\_category\` Nullable(Int64),  
\`platform\` Nullable(Int64),  
\`merchant\_id\` Nullable(String),  
\`merchant\_name\` Nullable(String),  
\`merchant\_img\` Nullable(String),  
\`order\_status\` Nullable(Int64),  
\`sub\_order\_status\` Nullable(Int64),  
\`sub\_order\_status\_str\` Nullable(String),  
\`total\_fee\` Nullable(Int64),  
\`wait\_fee\` Nullable(Int64),  
\`discount\_fee\` Nullable(Int64),  
\`real\_fee\` Nullable(Int64),  
\`pay\_status\` Nullable(Int64),  
\`pay\_type\` Nullable(Int64),  
\`pay\_id\` Nullable(String),  
\`pay\_no\` Nullable(String),  
\`pay\_time\` Nullable(String),  
\`is\_invoice\` Nullable(Int64),  
\`refund\_status\` Nullable(Int64),  
\`comment\_status\` Nullable(Int64),  
\`device\_platform\` Nullable(String),  
\`create\_time\` Nullable(String),  
\`update\_time\` Nullable(String),  
\`update\_version\` Nullable(Int64),  
\`is\_delete\` Nullable(Int64),  
\`is\_pre\` Nullable(Int64),  
\`cancel\_type\` Nullable(Int64),  
\`settle\_fee\_status\` Nullable(Int64),  
\`amount\` Nullable(Int64),  
\`discount\_amount\` Nullable(Int64),  
\`coupon\_discount\` Nullable(Int64),  
\`coupon\_create\_time\` Nullable(String),  
\`coupon\_promotion\_id\` Nullable(String),  
\`coupon\_promotion\_name\` Nullable(String),  
\`coupon\_type\` Nullable(Int64),  
\`coupon\_title\` Nullable(String),  
\`car\_id\` Nullable(String),  
\`car\_gear\` Nullable(String),  
\`car\_name\` Nullable(String),  
\`car\_seats\` Nullable(Int64),  
\`car\_emission\` Nullable(String),  
\`fetch\_address\` Nullable(String),  
\`return\_address\` Nullable(String),  
\`fetch\_time\` Nullable(String),  
\`return\_time\` Nullable(String),  
\`ckbizdate\` Date,  
\`bizdate\` Nullable(String)  
)ENGINE = MergeTree(ckbizdate, order\_id, 8192);

CREATE TABLE cdm\_dwd.dwd\_ord\_car\_sharing\_df\_cls on cluster crm\_4shards\_1replicas (  
\`order\_id\` String,  
\`p\_order\_id\` Nullable(String),  
\`order\_no\` Nullable(String),  
\`merchant\_order\_no\` Nullable(String),  
\`title\` Nullable(String),  
\`user\_no\` Nullable(String),  
\`order\_category\` Nullable(Int64),  
\`platform\` Nullable(Int64),  
\`merchant\_id\` Nullable(String),  
\`merchant\_name\` Nullable(String),  
\`merchant\_img\` Nullable(String),  
\`order\_status\` Nullable(Int64),  
\`sub\_order\_status\` Nullable(Int64),  
\`sub\_order\_status\_str\` Nullable(String),  
\`total\_fee\` Nullable(Int64),  
\`wait\_fee\` Nullable(Int64),  
\`discount\_fee\` Nullable(Int64),  
\`real\_fee\` Nullable(Int64),  
\`pay\_status\` Nullable(Int64),  
\`pay\_type\` Nullable(Int64),  
\`pay\_id\` Nullable(String),  
\`pay\_no\` Nullable(String),  
\`pay\_time\` Nullable(String),  
\`is\_invoice\` Nullable(Int64),  
\`refund\_status\` Nullable(Int64),  
\`comment\_status\` Nullable(Int64),  
\`device\_platform\` Nullable(String),  
\`create\_time\` Nullable(String),  
\`update\_time\` Nullable(String),  
\`update\_version\` Nullable(Int64),  
\`is\_delete\` Nullable(Int64),  
\`is\_pre\` Nullable(Int64),  
\`cancel\_type\` Nullable(Int64),  
\`settle\_fee\_status\` Nullable(Int64),  
\`amount\` Nullable(Int64),  
\`discount\_amount\` Nullable(Int64),  
\`coupon\_discount\` Nullable(Int64),  
\`coupon\_create\_time\` Nullable(String),  
\`coupon\_promotion\_id\` Nullable(String),  
\`coupon\_promotion\_name\` Nullable(String),  
\`coupon\_type\` Nullable(Int64),  
\`coupon\_title\` Nullable(String),  
\`car\_id\` Nullable(String),  
\`car\_gear\` Nullable(String),  
\`car\_name\` Nullable(String),  
\`car\_seats\` Nullable(Int64),  
\`car\_emission\` Nullable(String),  
\`fetch\_address\` Nullable(String),  
\`return\_address\` Nullable(String),  
\`fetch\_time\` Nullable(String),  
\`return\_time\` Nullable(String),  
\`ckbizdate\` Date,  
\`bizdate\` Nullable(String)  
)ENGINE = Distributed(crm\_4shards\_1replicas,cdm\_dwd ,dwd\_ord\_car\_sharing\_df,rand());

========================配置信息 dwd\_ord\_car\_sharing\_df\_ck.conf ==================  
\-- 编译配置信息gtdata04上新增文件 /home/admin/waterdrop/config/dwd\_ord\_car\_sharing\_df\_ck.conf  
  
spark \{  
[spark.app.name][] = "dwd\_ord\_car\_sharing\_df\_ck\_2\_ck"  
spark.executor.instances = 6  
spark.executor.cores = 2  
spark.executor.memory = "2g"  
spark.sql.catalogImplementation = "hive"  
\}

input \{  
hive \{  
pre\_sql = "select \* from cdm\_dwd.dwd\_ord\_car\_sharing\_df\_ck"  
table\_name = "dwd\_ord\_car\_sharing\_df\_ck"  
\}  
\}

filter \{  
remove \{  
\# source\_field = \["bizdate"\]  
\}  
  
convert \{  
source\_field = "order\_category"  
new\_type = "long"  
\}  
convert \{  
source\_field = "platform"  
new\_type = "long"  
\}  
convert \{  
source\_field = "order\_status"  
new\_type = "long"  
\}  
convert \{  
source\_field = "sub\_order\_status"  
new\_type = "long"  
\}  
convert \{  
source\_field = "total\_fee"  
new\_type = "long"  
\}  
convert \{  
source\_field = "wait\_fee"  
new\_type = "long"  
\}  
convert \{  
source\_field = "discount\_fee"  
new\_type = "long"  
\}  
convert \{  
source\_field = "real\_fee"  
new\_type = "long"  
\}  
convert \{  
source\_field = "pay\_status"  
new\_type = "long"  
\}  
convert \{  
source\_field = "pay\_type"  
new\_type = "long"  
\}  
convert \{  
source\_field = "is\_invoice"  
new\_type = "long"  
\}  
convert \{  
source\_field = "refund\_status"  
new\_type = "long"  
\}  
convert \{  
source\_field = "comment\_status"  
new\_type = "long"  
\}  
convert \{  
source\_field = "update\_version"  
new\_type = "long"  
\}  
convert \{  
source\_field = "is\_delete"  
new\_type = "long"  
\}  
convert \{  
source\_field = "is\_pre"  
new\_type = "long"  
\}  
convert \{  
source\_field = "cancel\_type"  
new\_type = "long"  
\}  
convert \{  
source\_field = "settle\_fee\_status"  
new\_type = "long"  
\}  
convert \{  
source\_field = "amount"  
new\_type = "long"  
\}  
convert \{  
source\_field = "discount\_amount"  
new\_type = "long"  
\}  
convert \{  
source\_field = "coupon\_discount"  
new\_type = "long"  
\}  
convert \{  
source\_field = "coupon\_type"  
new\_type = "long"  
\}  
convert \{  
source\_field = "car\_seats"  
new\_type = "long"  
\}

output \{  
clickhouse \{  
host = "10.2.12.56:8123"  
database = "cdm\_dwd"  
table = "cdm\_dwd.dwd\_ord\_car\_sharing\_df\_cls"  
username = "root"  
password = "root"  
\}  
\}  
  
  
\--第三步，导入数据(easyschedule添加调度任务)：

增量7天  
ssh dn1 clickhouse-client --query="alter table cdm\_dwd.dwd\_ord\_car\_sharing\_df delete where ckbizdate ='\\'$\{ckbizdate\}\\'' " ';  
ssh dn2 clickhouse-client --query="alter table cdm\_dwd.dwd\_ord\_car\_sharing\_df delete where ckbizdate ='\\'$\{ckbizdate\}\\'' " ';  
ssh dn3 clickhouse-client --query="alter table cdm\_dwd.dwd\_ord\_car\_sharing\_df delete where ckbizdate ='\\'$\{ckbizdate\}\\'' " ';  
ssh dn4 clickhouse-client --query="alter table cdm\_dwd.dwd\_ord\_car\_sharing\_df delete where ckbizdate ='\\'$\{ckbizdate\}\\'' " ';  
全量  
ssh dn1 clickhouse-client --query="truncate table cdm\_dwd.dwd\_ord\_car\_sharing\_df on cluster crm\_4shards\_1replicas" ';  
导入例子如下（也可以看这配置 http://212.64.114.87:8888/\#/projects/definition/list/440）

ifconfig
    
    ssh dn1 'clickhouse-client --query="truncate table cdm_dwd.dwd_ord_car_sharing_df on cluster crm_4shards_1replicas" ';
    ssh dn1 'echo `date` >> /tmp/wpp.log '
    sleep 1m
    
    beeline -u jdbc:hive2://gtdata03:10000 -n admin -p admin hive -hivevar ckbizdate=${ckbizdate} -f dwd_ord_car_sharing_df_ck.hql
    
    ssh gtdata04 '/home/admin/waterdrop/bin/start-waterdrop.sh --config /home/admin/waterdrop/config/dwd_ord_car_sharing_df_ck.conf -e client -m "local[6]"'

[spark.app.name]: http://spark.app.name/