北京制作公司网站/seo优化的基本流程

场 景

  紧接上一篇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 dataframereturn 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_sparkspark_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数据到mysqlspark_df.write.mode(mode)\.format("jdbc")\.options(url=url,user=user_name,password=password,dbtable=table_name,batchsize=batchsize,)\.save() #将从#数据全量插入表后，判断是否要执行善后的SQLif 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_sparkspark_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 serverspark_df.write.mode(mode)\.format("jdbc")\.options(url=url,user=user_name,password=password,dbtable=table_name,batchsize=batchsize,)\.save()#数据全量插入表后，判断是否要执行善后SQLif 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 dingdef 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_pathconfig.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 connexcept Exception as e:print(e)n = [0,97]ding.main(n)def main(argv):dataset_id = argv[1] #指定你在配置表内的数据集idgroup_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 serverexport_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++));
domysource_path=`eval echo '$'"source_path$i"`# echo ${mysource_path}hadoop fs -test -e "$mysource_path"hdfs_flag=$?if [ $hdfs_flag -eq 0 ];thenecho "mysource_path$i: $mysource_path hdfs file_data is exist!"elseecho "mysource_path$i: $mysource_path hdfs file_data is not exist"fi
done
#判断Source_path路径下文件是否都存在
function testfile()
{local total_success_flag=0for((i=1;i<=1;i++)); #需要修改domysource_path=`eval echo '$'"source_path$i"`hadoop fs -test -e $mysource_pathhdfs_flag=$?((total_success_flag+=hdfs_flag))doneecho $total_success_flag
}total_success_result=$(testfile)
echo "total not exists files: $total_success_result"
n=0
while (($total_success_result != 0))
dosleep 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 134exit 7fitotal_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 ]; thenecho "OK"else echo "failed"cd ${cur_dir}/../src/dw_distribute_main_py/ding_talk_warning_report_py/main/python3 ding_talk_with_agency.py 133exit 9
fi

  linux上调用shell

#20200416 日期
#配置表的dataset_id=11，group_id=2
#pushdown_department.sql hive取数脚本
#预处理脚本pre_pushdown_department.sql
#善后工作的脚本succeed_pushdown_department.sql
#重分区列dept_idbash 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上，当然这也不是一劳永逸的，遇到数据倾斜，目标库读写瓶颈等等，还是要根据具体问题进行调优！