Hadoop集群pyspark导出Hive集群数据及python写入mysql或sql server
zfyouxi
阅读：1475
2022-08-21 18:26:48
评论：0
场 景
紧接上一篇Hadoop集群数据分发——pyspark导出及python写入excel文件或csv文件及邮件附件发送，讲述了如何实现利用pyspark导出Hive集群数据到excel文件或csv文件，再以文件附件邮件发送，但是由于Hive内的数据本身对报表的展示，App的运用，主流BI工具分析都不是很好的兼容，所以很多情况下还需要把Hive的数据搬运到应用层，应用层可以指定一个MySQL或者SQL Server库，这里就讲一下如何利用pyspark将Hive的数据入到MySQL或SQL Server。
安装pyspark
一样，先安装好pyspark，装好的直接跳过，windows或者shell环境下,包有点大，可能有网络不稳定的话就多试几次，总会成功的，如果还不成功，再百度下办法。
pip install pyspark
总体文件夹
pushdown_to_mysql_or_sql_server/ # 总文件夹
├ pre_pushdown_department.sql # 预处理sql文件
├ pushdown_department.sql # hive上的取数脚本
├ pushdown_department.sh # 调用主函数的shell脚本
├ succeed_pushdown_department.sql # 善后的sql脚本
├ dw_distribute_main_py/ # 导入mysql，sql server的python主函数
│ ├ dw_distribute_database_main.py #主函数
│ ├ export_database.py #导入mysql或者sql server
│ ├ init_spark.py #初始化hive取数环境并得到取数结果data frame
│ ├ conf/ #配置文件
│ └ myconfig.ini
│ └ ding_talk_warning_report_py/ #参考之前我的钉钉报警博客
└ ……
定义Hive的pyspark取数脚本init_spark.py
首先第一步是利用pyspark去数据仓库Hive里面把数据取出来，存在dataframe上，注意 ding.main(n)是有异常的话钉钉报警，可以参考： 调度Job报错或异常触发钉钉报警（Python 3.x版）。
#-*-coding:utf-8-*-
from pyspark.sql import HiveContext,SparkSession
from pyspark.sql.functions import col
import os
import sys
from ding_talk_warning_report_py.main import ding_talk_with_agency as ding
#传入的参数是hive的取数脚本，确保脚本准确
def exec_hive(hivesql):
try:
#初始化，如果要执行hive sql,需要在初始化中加入enableHiveSupport()
spark = SparkSession.builder.master("yarn").appName("DistributeData").enableHiveSupport().getOrCreate()
hive_context = HiveContext(spark)#初始化pyspark的环境变量
spark_df = hive_context.sql(hivesql)#执行Hive sql结果赋值给spark dataframe
return spark_df #返回一个spark的dataframe，dataframe可以理解为一张表
except Exception as e:
print(e)
n = [0,94]
ding.main(n) #报错的话钉钉报警，可以参考：调度Job报错或异常触发钉钉报警（Python 3.x版）
raise e //异常外抛，系统爆错出来
将spark dataframe的结果导入MySQL或者SQL Server的export_database.py
要导入到MySQL或者SQL Server之前需要准备安装相应的Python包，如下：
pip install pymysql #安装mysql的依赖包
pip install pymssql #安装sql server依赖的包
对pyspark的基础理解可以参考官网文档，本质还是调用了JDBC来操作数据库，官方文档给出的pyspark利用JDBC读写数据库样例：
# Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
# Loading data from a JDBC source
jdbcDF = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql:dbserver") \
.option("dbtable", "schema.tablename") \
.option("user", "username") \
.option("password", "password") \
.load()
jdbcDF2 = spark.read \
.jdbc("jdbc:postgresql:dbserver", "schema.tablename",
properties={
"user": "username", "password": "password"})
# Specifying dataframe column data types on read
jdbcDF3 = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql:dbserver") \
.option("dbtable", "schema.tablename") \
.option("user", "username") \
.option("password", "password") \
.option("customSchema", "id DECIMAL(38, 0), name STRING") \
.load()
# Saving data to a JDBC source
jdbcDF.write \
.format("jdbc") \
.option("url", "jdbc:postgresql:dbserver") \
.option("dbtable", "schema.tablename") \
.option("user", "username") \
.option("password", "password") \
.save()
jdbcDF2.write \
.jdbc("jdbc:postgresql:dbserver", "schema.tablename",
properties={
"user": "username", "password": "password"})
# Specifying create table column data types on write
jdbcDF.write \
.option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)") \
.jdbc("jdbc:postgresql:dbserver", "schema.tablename",
properties={
"user": "username", "password": "password"})
接下来是如何写入MySQL和SQL Server的文件export_database.py明细:
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# vim:fenc=utf-8
from pyspark.sql import HiveContext,SparkSession
import init_spark
import pymysql
import pymssql
import os
import sys
from sqlalchemy import create_engine
from ding_talk_warning_report_py.main import ding_talk_with_agency as ding
#定义写入mysql的函数export_mysql
#hive_sql 在hive上执行的取数脚本
#mode写入mysql或者sql server的模式，
#mode支持overwrite,会根据字段类型自动建表，类似sql server的select * into tablea from tableb的用法，不推荐，有的时候字段类型拿捏不准。
#mode支持append，在原表的基础上追加数据，推荐。
#url数据库连接，mysql一定要指定serverTimezone，不然时间会错乱，如：jdbc:mysql://10.245.628.631:3306/datacenter?serverTimezone=GMT%2B8&characterEncoding=UTF-8
#jdbc:sqlserver://10.216.232.35:1433;DatabaseName=DB_FWO2_DW
#user_name,password,table_name 用户名，密码，表名
#delete_sql预处理语句，比如写今天数据进表前需要先删除今天数据容错，写入表前需要先建表等，支持多条，用";"隔开，后面有循环会处理
#success_sql写入数据到表内后做的善后工作，比如写入日志表告诉下游我好了，给表改名等，支持多条，每一句用";"结尾隔开，最后一句也要，后面有循环会处理
#batchsize每次体检的到数据库的条数，可以作为优化参数，但是感觉效果不大
#my_partition_key是否要对hive的结果进行重分区，这个优化很重要
def export_mysql(hive_sql,mode,url,user_name,password,table_name,delete_sql='',success_sql='',batchsize="20000",my_partition_key=''):
try:
#生成连接指定mysql的连接信息
conn_str="mysql+pymysql://%s:%s@%s?charset=utf8"%(user_name,password,url[13:url.index("?")])
db_engine=create_engine(conn_str,echo=False,encoding="utf-8")
#调用init_spark
spark_df = init_spark.exec_hive(hive_sql)
print("分区列为:")
print(my_partition_key)
#如果遇到从Hive读取出来的数据数据倾斜，即最终都集中到一个task写入mysql
#数据显然就很慢，所以这时就要对hive得到的data frame重分区，这里的8是分为8个，可以跟自己的数据情况来设置，一般是自己executor*cores 的数量的一倍或两倍，我设置--executor-cores 2 --num-executors 4，所以写8
#效果一个180M的文件没重分区前，数据倾斜写入耗时30分钟，重分区后，7分钟，注意重分区本身耗时，因为你要把原来的整体数据再打散或者合并，打散还会触发shuffle操作。
spark_df = spark_df.repartition(8,my_partition_key)
#overwrite重新建表方式
if mode == "overwrite":
#isolationLevel是否开启事务隔离，该参数优化感觉效果也不佳
spark_df.write.mode(mode)\
.format("jdbc")\
.options(url=url,user=user_name,password=password,dbtable=table_name,batchsize=batchsize,isolationLevel="NONE")\
.save()
# append 追加方式
elif mode == "append":
if delete_sql:
#如果有预处理语句，执行预处理语句，用分号分割获取多条
sql_list = delete_sql.split(';')[:-1]
for x in sql_list:
# 判断包含空行的
if '\n' in x:
# 替换空行为1个空格
x = x.replace('\n', ' ')
# 判断多个空格时
if ' ' in x:
# 替换为空
x = x.replace(' ', '')
# sql语句添加分号结尾
sql_item = x+';'
# print(sql_item)
db_engine.execute(sql_item) #循环体内执行每一条预处理语句
print("执行成功sql: %s"%sql_item)
#spark insert数据到mysql
spark_df.write.mode(mode)\
.format("jdbc")\
.options(url=url,user=user_name,password=password,dbtable=table_name,batchsize=batchsize,)\
.save() #将从
#数据全量插入表后，判断是否要执行善后的SQL
if success_sql:
print("start to run success_sql")
sql_list = success_sql.split(';')[:-1]
for x in sql_list:
# 判断包含空行的
if '\n' in x:
# 替换空行为1个空格
x = x.replace('\n', ' ')
# 判断多个空格时
if ' ' in x:
# 替换为空
x = x.replace(' ', '')
# sql语句添加分号结尾
sql_item = x+';'
# print(sql_item)
#循环体内执行多条善后的语句
db_engine.execute(sql_item)
print("执行成功sql: %s"%sql_item)
except Exception as e:
print(e)
print('执行失败sql: %s'%sql_item)
raise e
finally:
db_engine.dispose()
#定义写入sql server的函数export_mysql
#hive_sql 在hive上执行的取数脚本
#mode写入mysql或者sql server的模式，
#mode支持overwrite,会根据字段类型自动建表，类似sql server的select * into tablea from tableb的用法，不推荐，有的时候字段类型拿捏不准。
#mode支持append，在原表的基础上追加数据，推荐。
#url数据库连接，mysql一定要指定serverTimezone，不然时间会错乱，如：jdbc:mysql://10.245.628.631:3306/datacenter?serverTimezone=GMT%2B8&characterEncoding=UTF-8
#jdbc:sqlserver://10.216.232.35:1433;DatabaseName=DB_FWO2_DW
#user_name,password,table_name 用户名，密码，表名
#delete_sql预处理语句，比如写今天数据进表前需要先删除今天数据容错，写入表前需要先建表等，支持多条，用";"隔开，后面有循环会处理
#success_sql写入数据到表内后做的善后工作，比如写入日志表告诉下游我好了，给表改名等，支持多条，每一句用";"结尾隔开，最后一句也要，后面有循环会处理
#batchsize每次体检的到数据库的条数，可以作为优化参数，但是感觉效果不大
#my_partition_key是否要对hive的结果进行重分区，这个优化很重要
def export_sqlserver(hive_sql,mode,url,user_name,password,table_name,delete_sql='',success_sql='',batchsize="1000",my_partition_key=''):
try:
#生成连接指定mysql的连接信息
conn_str="mssql+pymssql://%s:%s@%s/%s?charset=utf8"%(user_name,password,url[17:url.index(";")],url[url.index("DatabaseName")+13:])
db_engine=create_engine(conn_str,isolation_level="AUTOCOMMIT",echo=False,encoding="utf-8")
#调用init_spark
spark_df = init_spark.exec_hive(hive_sql)
print("分区列为:")
print(my_partition_key)
#如果遇到从Hive读取出来的数据数据倾斜，即最终都集中到一个task写入mysql
#数据显然就很慢，所以这时就要对hive得到的data frame重分区，这里的8是分为8个，可以跟自己的数据情况来设置，一般是自己executor*cores 的数量的一倍或两倍，我设置--executor-cores 2 --num-executors 4，所以写8
#效果一个180M的文件没重分区前，数据倾斜写入耗时30分钟，重分区后，7分钟，注意重分区本身耗时，因为你要把原来的整体数据再打散或者合并，打散还会触发shuffle操作。
spark_df = spark_df.repartition(8,my_partition_key)
#overwrite重新建表方式
#overwrite重新建表方式
if mode == "overwrite":
spark_df.write.mode(mode)\
.format("jdbc")\
.options(url=url,user=user_name,password=password,dbtable=table_name,batchsize=batchsize,)\
.save()
# append 追加方式
elif mode == "append":
if delete_sql:
#执行预处理语句
sql_list = delete_sql.split(';')[:-1]
for x in sql_list:
# 判断包含空行的
if '\n' in x:
# 替换空行为1个空格
x = x.replace('\n', ' ')
# 判断多个空格时
if ' ' in x:
# 替换为空
x = x.replace(' ', '')
# sql语句添加分号结尾
sql_item = x+';'
# print(sql_item)
#循环执行遍历每一句
db_engine.execute(sql_item)
print("执行成功sql: %s"%sql_item)
#spark insert数据到sql server
spark_df.write.mode(mode)\
.format("jdbc")\
.options(url=url,user=user_name,password=password,dbtable=table_name,batchsize=batchsize,)\
.save()
#数据全量插入表后，判断是否要执行善后SQL
if success_sql:
sql_list = success_sql.split(';')[:-1]
for x in sql_list:
# 判断包含空行的
if '\n' in x:
# 替换空行为1个空格
x = x.replace('\n', ' ')
# 判断多个空格时
if ' ' in x:
# 替换为空
x = x.replace(' ', '')
# sql语句添加分号结尾
sql_item = x+';'
# print(sql_item)
db_engine.execute(sql_item)
print("执行成功sql: %s"%sql_item)
except Exception as e:
print(e)
print('执行失败sql: %s'%sql_item)
raise e
finally:
db_engine.dispose()
主函数调用dw_distribute_database_main.py
主函数涉及到数据库里面的配置文件，所以首先在数据库里面建个表，主要用来存储每次一导入数据库用到的配置文件，cfg_dw_distribute_database_dataset建表语句如下：
CREATE TABLE `cfg_dw_distribute_database_dataset` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`dataset_id` bigint(20) DEFAULT NULL COMMENT '数据集id',
`dataset_name` varchar(50) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '数据集名称',
`group_id` bigint(20) DEFAULT NULL COMMENT 'dataset下组id',
`sequenceid` int(11) DEFAULT NULL COMMENT 'dataset下组的排序id',
`db_type` varchar(20) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '数据库类型，目前支持两种:mysql,mssql',
`mode` varchar(20) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '写入数据库的模式：overwrite,append',
`db_url` varchar(500) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '数据库连接地址：jdbc:sqlserver://host:port;DatabaseName=db_name或jdbc:mysql://host:port/db_name',
`db_user` varchar(20) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '用户名',
`db_password` varchar(20) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '密码',
`target_table` varchar(100) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '目标表名',
`hive_sql_name` varchar(100) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '查询hive的SQL语句的文件名称，默认文件在/src/目录下',
`delete_sql_name` varchar(100) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '删除目标表的数据的SQL文件名称，默认文件在/src/目录下，如果没有删除语句，该列留空',
`success_sql_name` varchar(100) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '数据插入后执行的SQL文件名称，默认文件在/src/目录下，如果没有执行语句，该列留空',
`is_enable` tinyint(4) DEFAULT NULL COMMENT '是否有效',
`creator` varchar(50) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '创建者',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uidx_cfg_dataset` (`dataset_id`,`group_id`,`target_table`)
) ENGINE=InnoDB AUTO_INCREMENT=26 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
接下来是用到的./conf/myconfig.in下记录的上表的登录信息到配置，如下
[mysql_dw_config]
host = 10.589.72.231
port = 3306
user = user
passwd = iloveyou123
db = dw_config
charset = utf8
主函数代码如下：
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# vim:fenc=utf-8
from pyspark.sql import HiveContext,SparkSession
from pyspark.sql.functions import col
import os
import configparser
import sys
import pandas as pd
import pymysql
import pymssql
import export_database
from ding_talk_warning_report_py.main import ding_talk_with_agency as ding
def getcon():
#读取配置文件信息，获取到配置表cfg_dw_distribute_database_dataset内你想要利用写到mysql和sql server的的配置
config = configparser.ConfigParser()
file_path = os.path.dirname(__file__)
ini_path = "%s/conf/myconfig.ini"%file_path
config.read(ini_path)
config_host = config['mysql_dw_config']["host"]
config_port = int(config['mysql_dw_config']["port"])
config_user = config['mysql_dw_config']["user"]
config_passwd = config['mysql_dw_config']["passwd"]
config_db = config['mysql_dw_config']["db"]
config_charset = config['mysql_dw_config']["charset"]
try:
conn = pymysql.Connect(host=config_host, port=config_port, database=config_db, user=config_user,password=config_passwd, charset=config_charset)
return conn
except Exception as e:
print(e)
n = [0,97]
ding.main(n)
def main(argv):
dataset_id = argv[1] #指定你在配置表内的数据集id
group_id = argv[2] #指定数据集的组id，和dataset_id一起能唯一确定一条数据分发
hive_sql = argv[3] #取数sql脚本的文本内容
del_sql = argv[4] #预处理sql脚本的内容
suc_sql = argv[5] #善后工作的sql脚本内容
partion_key= argv[6] #分区列名
#获取配置表数据库连接信息
conn=getcon()
#得到写入mysql或者sql server的配置参数
sqlcmd = """ SELECT *
FROM cfg_dw_distribute_database_dataset
WHERE dataset_id = %s and group_id = %s and is_enable=1
"""%(dataset_id,group_id)
config_df = pd.read_sql(sqlcmd,conn)
if config_df.empty:
print("no the dateset and group infomation")
else:
for index,row in config_df.iterrows():
db_type = row['db_type']
mode = row['mode']
db_url = row['db_url']
db_user = row['db_user']
db_password = row['db_password']
target_table = row['target_table']
#调用写入mysql或者sql server的python脚本
if db_type == 'mysql':
#调用export_database.py中的export_mysql函数，把数据导入mysql中
export_database.export_mysql(hive_sql=hive_sql,mode=mode,url=db_url,user_name=db_user,password=db_password,table_name=target_table,delete_sql=del_sql,success_sql=suc_sql,my_partition_key=partion_key)
elif db_type == 'mssql':
#调用SQL server
export_database.export_sqlserver(hive_sql=hive_sql,mode=mode,url=db_url,user_name=db_user,password=db_password,table_name=target_table,delete_sql=del_sql,success_sql=suc_sql)
conn.close()
if __name__ == "__main__":
main(sys.argv)
调用shell脚本调度主函数shell脚本pushdown_department.sh：
#!/usr/bin/env bash
#sh
mydate=${1}
dataset_id=${2}
group_id=${3}
mydate=$(date -d"1 day ago ${mydate}" +%Y%m%d) #获取昨天
pre_one_month_day=$(date -d"1 month ago ${mydate}" +%Y%m%d) #获取上个月
event_week=$(date -d ${
mydate} +%V)
event_week=$((10#$event_week)) #convert enentweek to Int type
event_hour='00'
hive_sql_file=${4}
delete_sql_file=${5}
success_sql_file=${6}
partition_key=${7}
cur_dir=`pwd`
hive_sql_file=`echo "${cur_dir}/../src/${hive_sql_file}"`
delete_sql_file=`echo "${cur_dir}/../src/${delete_sql_file}"`
success_sql_file=`echo "${cur_dir}/../src/${success_sql_file}"`
#hive sql,如果有参数替换脚本内的参数
hive_sql=`cat ${
hive_sql_file} | tr "\n" " " | tr "\r" " "`
hive_sql=`printf "${hive_sql}" "${mydate}"` #替换SQL文件中的%s,%d,%f变量值
#delete_sql，如果有参数替换脚本内的参数
delete_sql=`cat ${
delete_sql_file} | tr "\n" " " | tr "\r" " "`
#delete_sql=`printf "${delete_sql}" "${mydate}" "${mydate}"` #替换SQL文件中的%s,%d,%f变量值
#success_sql，如果有参数替换脚本内的参数
success_sql=`cat ${
success_sql_file} | tr "\n" " " | tr "\r" " "`
success_sql=`printf "${success_sql}"` #替换SQL文件中的%s,%d,%f变量值
echo "mydate: $mydate"
echo "pre_one_month_day: $pre_one_month_day"
echo "event_week: $event_week"
echo "cur_dir: $cur_dir"
echo "hive_sql: $hive_sql"
echo "delete_sql: $delete_sql"
echo "success_sql: $success_sql"
echo "partition_key: $partition_key"
#source_path1 查看你依赖的表在Hive是否ready
source_path1=/hive/warehouse/ods/rs/basic/ods_rs_basic_tbb_department/event_week=${event_week}/event_day=${mydate}/event_hour=${event_hour}/_SUCCESS
#输出Source_path路径下文件是否都存在，用循环时我可能有多个依赖文件，所以我懒得改了
for((i=1;i<=1;i++));
do
mysource_path=`eval echo '$'"source_path$i"`
# echo ${mysource_path}
hadoop fs -test -e "$mysource_path"
hdfs_flag=$?
if [ $hdfs_flag -eq 0 ];then
echo "mysource_path$i: $mysource_path hdfs file_data is exist!"
else
echo "mysource_path$i: $mysource_path hdfs file_data is not exist"
fi
done
#判断Source_path路径下文件是否都存在
function testfile()
{
local total_success_flag=0
for((i=1;i<=1;i++)); #需要修改
do
mysource_path=`eval echo '$'"source_path$i"`
hadoop fs -test -e $mysource_path
hdfs_flag=$?
((total_success_flag+=hdfs_flag))
done
echo $total_success_flag
}
total_success_result=$(testfile)
echo "total not exists files: $total_success_result"
n=0
while (($total_success_result != 0))
do
sleep 1m
((n++))
if [ $n -gt 60 ]; then
echo "等待1个小时，系统数据还没加载到hdfs"
python3 ${cur_dir}/../src/dw_distribute_main_py/ding_talk_warning_report_py/main/ding_talk_with_agency.py 134
exit 7
fi
total_success_result=$(testfile)
echo "total not exists files: $total_success_result"
done
#调用python脚本，注意这个很重要，采用spark-submit 提交，而不是直接puthon 执行
spark-submit --master yarn --deploy-mode client --executor-memory 4G --executor-cores 2 --num-executors 4 --queue etl ${cur_dir}/../src/dw_distribute_main_py/dw_distribute_database_main.py ${dataset_id} ${group_id} "${hive_sql}" "${delete_sql}" "${success_sql}" "${partition_key}"
status=$?
if [ $status == 0 ]; then
echo "OK"
else
echo "failed"
cd ${cur_dir}/../src/dw_distribute_main_py/ding_talk_warning_report_py/main/
python3 ding_talk_with_agency.py 133
exit 9
fi
linux上调用shell
#20200416 日期
#配置表的dataset_id=11，group_id=2
#pushdown_department.sql hive取数脚本
#预处理脚本pre_pushdown_department.sql
#善后工作的脚本succeed_pushdown_department.sql
#重分区列dept_id
bash pushdown_department.sh 20200416 11 2 pushdown_department.sql pre_pushdown_department.sql succeed_pushdown_department.sql dept_id
pushdown_department.sql样例，无分号结尾:
select dept_id, dept_name, dept_no, k3_dept_no, manager, dept_level, city_id, parent_dept_id, status, k3_dept_name, fk3_dept_no, fk3_dept_name, dept_funs, dept_type, event_week, event_day, event_hour
,`current_timestamp`() as Load_time
from dw.ods_rs_basic_tbb_department
where event_day='%s'
pre_pushdown_department.sql样例,有分号结尾:
INSERT INTO publish.data_check_log (insert_time, status, remarks, check_table)
VALUES (now(), '202', 'processing', 'stg_rs_basic_tbb_department');
truncate table stg_rs_basic_tbb_department;
succeed_pushdown_department.sql样例,有分号结尾:
INSERT INTO publish.data_check_log (insert_time, status, remarks, check_table)
VALUES (now(), '200', 'successful', 'stg_rs_basic_tbb_department');
到此就结束了，可以愉快的将Hive上的数据交互到mysql或者sql server上，当然这也不是一劳永逸的，遇到数据倾斜，目标库读写瓶颈等等，还是要根据具体问题进行调优！
本文参考链接：https://blog.csdn.net/LXWalaz1s1s/article/details/105207239
声明
1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，请转载时务必注明文章作者和来源，不尊重原创的行为我们将追究责任；3.作者投稿可能会经我们编辑修改或补充。
