Shlle脚本传参调用seatunnel(原waterdrop)将hive中数据导入ClickHouse
前言
公司分析数据已经存入hive,但需要输入参数计算得到很长一段时间的趋势变化数据(不固定查询),经调研ClickHouse时序优化后比较满足需求,并且ClickHouse在数据量大时最好采用DNS轮询本地表写,分布式表读的工作方式。
这里测试写脚本将hive中数据导入ClickHouse,加入到原来的数仓流程。
其实可以采取kafka+spark/streaming方式批量插入clickhouse提供准实时计算,后续看需求吧
开始测试:
运行环境
首先,假设已经安装好seatunnel1.5.1(waterdrop1.5.1),并且安装spark2.4.8、scala(这三个搭配是官方的推荐的),$HADOOP_CONF/hive-site.xml放入spark2.4.8/conf。
百度网盘自取:
链接:https://pan.baidu.com/s/1BZ8-oNXhRjmrqd3hW-KxXA 提取码:hevt
环境变量:我这里hive3.1.2使用的是Spark3,所以配了一个SPARK2_HOME,后续脚本中设置使用SPARK2_HOME
# SPARK_HOME
export SPARK_HOME=/u/module/spark
export PATH=$PATH:$SPARK_HOME/bin
# SPARK_END# 多版本共存Spark,for waterdrop and Hive
export SPARK2_HOME=/u/module/spark-2.4.8-bin-hadoop2.7#Scala Env
export SCALA_HOME=/u/module/scala-2.11.8/
export PATH=$PATH:$SCALA_HOME/bin
创建jobs目录存放执行conf文件
mkdir /u/module/seatunnel-1.5.1/jobs
创建表
hive:
-- auto-generated definition
create table student
(id int,name string
)row format serde 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'stored asinputformat 'org.apache.hadoop.mapred.TextInputFormat'outputformat 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'location 'hdfs://hadoop101:8020/user/hive/warehouse/student'tblproperties ('bucketing_version' = '2');
插入:
insert into student values (1,'abc'),(2,'def');
clickhouse:
drop table if exists waterdrop_test;
create table waterdrop_test
(id UInt16,name String,birthday Date
) ENGINE = TinyLog;
脚本
我们可以使用cat <<!EOF把变量传进去,并且把脚本生成在文件夹中,然后再使用waterdrop的命令调用即可
touch ~/bin/mytest.sh && chmod u+x ~/bin/mytest.sh && vim ~/bin/mytest.sh
#!/bin/bash# 环境变量
unset SPARK_HOME
export SPARK_HOME=$SPARK2_HOME
SEATUNNEL_HOME=/u/module/seatunnel-1.5.1
# 接收两个参数,第一个为要抽取的表,第二个为抽取时间
# 若输入的第一个值为first,不输入第二参数则直接退出脚本
if [[ $1 = first ]]; thenif [ -n "$2" ] ;thendo_date=$2else echo "请传入日期参数"exitfi
# 若输入的第一个值为all,不输入第二参数则取前一天
elif [[ $1 = all ]]; then# 判断非空,如果不传时间默认取前一天数据,传时间就取设定,主要是用于手动传参if [ -n "$2" ] ;thendo_date=$2elsedo_date=`date -d '-1 day' +%F`fi
elseif [ -n "$2" ] ;thendo_date=$2else echo "请传入日期参数"exitfi
fiecho "日期:$do_date"# 打印数据传输脚本并赋值
cat>$SEATUNNEL_HOME/jobs/hive2ck_test.conf<<!EOF
spark {spark.sql.catalogImplementation = "hive"spark.app.name = "hive2clickhouse"spark.executor.instances = 2spark.executor.cores = 2spark.executor.memory = "2g"
}input {hive {pre_sql = "select id,name,'${do_date}' as birthday from default.student"table_name = "test"}
}filter {}output {clickhouse {host = "hadoop101:8123"database = "default"table = "waterdrop_test"fields = ["id","name","birthday"]username = "default"password = ""}
}!EOF$SEATUNNEL_HOME/bin/start-waterdrop.sh --config $SEATUNNEL_HOME/jobs/hive2ck_test.conf -e client -m 'local[2]'
进阶脚本
#!/bin/bash# 环境变量
unset SPARK_HOME
export SPARK_HOME=$SPARK2_HOME
SEATUNNEL_HOME=/u/module/seatunnel-1.5.1
# 接收两个参数,第一个为要抽取的表,第二个为抽取时间
# 若输入的第一个值为first,不输入第二参数则直接退出脚本
if [[ $1 = first ]]; thenif [ -n "$2" ] ;thendo_date=$2else echo "请传入日期参数"exitfi
# 若输入的第一个值为all,不输入第二参数则取前一天
elif [[ $1 = all ]]; then# 判断非空,如果不传时间默认取前一天数据,传时间就取设定,主要是用于手动传参if [ -n "$2" ] ;thendo_date=$2elsedo_date=`date -d '-1 day' +%F`fi
elseif [ -n "$2" ] ;thendo_date=$2else echo "请传入日期参数"exitfi
fiecho "日期:$do_date"import_conf(){# 打印数据传输脚本并赋值
cat>$SEATUNNEL_HOME/jobs/hive2ck_test.conf<<!EOF
spark {spark.sql.catalogImplementation = "hive"spark.app.name = "hive2clickhouse"spark.executor.instances = 2spark.executor.cores = 4spark.executor.memory = "4g"
}input {hive {# pre_sql = "select id,name,'${do_date}' as birthday from default.student"pre_sql = "$1"table_name = "$2"}
}filter {}output {clickhouse {host = "$3"database = "$4"table = "$5"# fields = ["id","name","birthday"]fields = $6username = "default"password = ""}
}!EOF
$SEATUNNEL_HOME/bin/start-waterdrop.sh --config $SEATUNNEL_HOME/jobs/hive2ck_test.conf -e client -m 'local[2]'
}import_test(){import_conf "select id,name,'${do_date}' as birthday from default.student" "test" "hadoop101:8123" "default" "waterdrop_test" "[\"id\",\"name\",\"birthday\"]"
}import_test
此脚本将生成conf做为一个方法,可用于多表导入
后续增量导入
总结
我这里测试成功,这只是最基本的使用,可根据自己需求更改。
若对你有帮助,点个赞吧~
Shlle脚本传参调用seatunnel(原waterdrop)将hive中数据导入ClickHouse相关推荐
- 【clickhouse】使用waterdrop将Hive中的数据导入ClickHouse
1.概述 转载:使用waterdrop将Hive中的数据导入ClickHouse 这里仅仅自己学习用. 前言 最近有一个需求需要把hive的数据同步到clickhouse,而且数据量还比较大,所以使用 ...
- c#直接调用ssis包实现Sql Server的数据导入功能
调用ssis包实现Sql Server的数据导入功能网上已经有很多人讨论过,自己参考后也动手实现了一下,上一次笔者的项目中还用了一下这个功能.思前想后,决定还是贴一下增强记忆,高手请54. 1.直接调 ...
- 在Java中动态传参调用Python脚本
最近,又接触到一个奇葩的接口,基于老板不断催促赶时间的情况下,在重写java接口和复用已有的python脚本的两条路中选择了后者,但是其实后者并没有好很多,因为我是一个对python的认识仅限于其名称 ...
- python脚本传参
在Linux环境调用python脚本,需要从外部传参,python提供了两种传参方式: 1.顺序传参 import sys 它封装了与python解释器相关的数据,在脚本里面使用使用参数的顺序必须和传 ...
- Golang Post传参调用示例
Golang Post传参 package mainimport ("fmt""net/http""io/ioutil""str ...
- 二维数组的传参调用写法
初遇二维数组作函数参数,宛如自己化身为了大头儿子.很头大. 不禁大声呐喊:该怎么声明定义,该怎么调用,又该怎么函数中操作元素和地址? 在此,我要拨开这些问题的一些迷雾. 我相信,有心人看完后,再遇就不 ...
- python中两个文件如何互相传参_argparse模块如何在jupyter notebook中用于传参?
作者:于晨晨 研究方向:nlp 发表于公众号:AI技术日常 在python代码文件中,通常需要传参,传参就需要经常使用argparse.使用argparse模块在py文件中是正常的,但是jupyter ...
- python如何次传参给线程_python如何给线程中的函数传参?
1.Process说明 (1)概念 process模块是一个创建进程的模块,借助这个模块,就可以完成进程的创建. (2)语法([group [, target [, name [, args [, k ...
- python调用bat脚本传参,将参数从批处理文件传递给Python
I am trying to pass argument from batch file to python as following. It seems nothing has passed to ...
最新文章
- JDK8:Lambda表达式操作List集合
- oracle10grac创建单实例,Oracle10gLinux单实例迁移到rac
- Python 中list.sort和sorted以及bisect
- data center!
- 计算机视觉实时目标检测 TensorFlow Object Detection API
- 图像的连通域检测的堆栈算法
- BO QUERY BUILDER - SI_INSTANCE相关属性
- 各种震撼的慢镜头,奇怪的知识又增加了!​
- 为啥你写的代码老有大串的if/else?
- python根据相关系数绘制热力图
- redis学习-03
- DjVu Reader Pro for Mac(DjVu阅读软件)
- vue项目小demo
- Lucene的分析过程
- html动态背景gif图片,gif动态背景
- 作为一只爬虫,如何科学有效地处理短信验证码?
- 测试自动化风险或成功的条款
- 基于redis实现活跃用户统计功能
- 2020年开发人员的机器学习基础
- 图形界面介绍Create Pin Blockage