DataX自动生成json配置文件及多线程执行脚本(仅支持mysql-->hive),其他版本自行实现,修改json模版即可
执行Datax任务
datax_run.py
# 指定项目下的所有的ods任务,如果不指定参数,默认执行dw下的prefix过滤后的所有抽取任务
# 使用方式:python3 datax_run.py -p 项目名 -f 过滤json的前缀或多个文件名,拼接
import os
import re
import sys
import json
import time
import argparse
import subprocess
from threading import Thread
# from logger import create_logger
# 必须在datax目录下执行
datax_home = None
datax = f"bin/datax.py"
datax_config_dir = f"job/dw"
logs_dir = "logs"
prefix = ["ods_"]
subfix = [".json"]
thread_num = 8
task_count = 0
def init():
global datax_home
# 检查配置及环境变量是否存在问题
# 必须在datax目录下执行
environ = os.environ
if not environ.keys().__contains__("DATAX_HOME"):
print("未找到环境变量[DATAX_HOME]")
return False
datax_home = environ.get("DATAX_HOME")
if datax_home is None:
print("环境变量[DATAX_HOME]未设置值")
return False
else:
hive)
datax_config_dir = f"job/{hive}"
logs_dir = f"{logs_dir}/{hive}"
if args_filter is not None:
print("过滤条件：", args_filter)
prefix.clear()
for config in args_filter.split(","):
prefix.append(config)
elif hive is not None:
prefix = ["ods_"]
print(f"初始化参数:配置路径--> {datax_config_dir}\nprefix--> {prefix}")
def run_sub_shell(cmd):
try:
# cmd = f"source /etc/profile && cd {datax_home} && " + cmd
# print(cmd)
output = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT, encoding="utf-8", cwd=datax_home)
print(f'Command "{cmd}" executed successfully. Output: \n{output}')
return output
except subprocess.CalledProcessError as e:
print(f'Command "{cmd}" failed with error: \n{e.output}')
exit(-1)
def check_status(log_path):
with open(file=log_path, mode="r", encoding="utf-8") as fp:
if fp.read().__contains__("completed succes print(f"datax_home:{datax_home}")
return True
if not init():
print("初始化失败,正在退出...")
exit(-1)
# log = create_logger("datax_run", f"{datax_home}/logs/")
def extract_filename(file_path):
re_search = re.search(r'([^\\/]+)\.json$', file_path)
if re_search:
return re_search.group(1)
else:
print("未匹配到文件名,请检查...")
return None
def check_args():
parser = argparse.ArgumentParser(description='datax批量任务执行脚本')
# 添加--hive参数
parser.add_argument("--hive", type=str, help="hive数据库")
# 添加-f/--filter参数
parser.add_argument("-f", "--filter", type=str, help="输入过滤条件")
# 解析命令行参数
args = parser.parse_args()
hive = args.hive
args_filter = args.filter
# 输出结果
global prefix, datax_config_dir, logs_dir
if hive is None:
print(f"默认使用配置目录[{datax_config_dir}]")
else:
print("目标Hive库：",sfully"):
print(f"执行datax任务成功:{log_path}")
return True
else:
print(f"执行datax任务失败:{log_path}")
exit(-1)
def init_log():
# 获取今天日期
from datetime import datetime
date_str = datetime.today().date()
# 创建目录
global logs_dir
logs_dir = f"{datax_home}/{logs_dir}/{date_str}"
os.makedirs(logs_dir, exist_ok=True)
print(f"logs dir[{logs_dir}]")
def match_config(x: str, prefix: [], subfix: []):
for pre in prefix:
if x.startswith(pre):
for sub in subfix:
if x.endswith(sub):
return True
return False
def thread_run(config):
config_name = extract_filename(config)
cmd = f"python {datax} {config}"
# cmd = f"python {datax} {config} > {logs_dir}/{config_name}.log"
output = run_sub_shell(cmd)
if output.__contains__("completed successfully"):
task_count -= 1
print(f"同步数据[{config_name}]成功,剩余任务{task_count}...")
else:
print(f"同步数据[{config_name}]失败!")
exit(-1)
def gen_thread_data():
full_paths = []
# 指定配置文件目录
for dirpath, dirnames, filenames in os.walk(f"{datax_home}/{datax_config_dir}"):
configs = filter(lambda x: match_config(x, prefix, subfix), filenames)
full_paths = [dirpath + "/" + x for x in configs]
return full_paths
def future_thread():
from concurrent import futures
thread_data = gen_thread_data()
global task_count
task_count = len(thread_data)
print(f"待执行抽取任务数量:{task_count}")
with futures.ThreadPoolExecutor(max_workers=thread_num) as executor:
for elem in thread_data:
executor.submit(thread_run, elem)
def start():
check_args()
# init_log()
future_thread()
if __name__ == '__main__':
start()生成Datax配置,创建ODS表
根据配置生成Datax配置json和自动创建hive ods表的脚本build_core.py
import json
import re
import sys
from pathlib import Path
import mysql.connector
from pyhive import hive
import os
import subprocess
datax_home = None
# 初始化检查环境
def init():
global datax_home
# 检查配置及环境变量是否存在问题
# 必须在datax目录下执行
environ = os.environ
if not environ.keys().__contains__("DATAX_HOME"):
print("未找到环境变量[DATAX_HOME]")
return False
datax_home = environ.get("DATAX_HOME")
if datax_home is None:
print("环境变量[DATAX_HOME]未设置值")
return False
else:
print(f"datax_home:{datax_home}")
return True
if not init():
print("初始化失败,正在退出...")
exit(-1)
# 主要内容用于生成datax的mysql到hive的配置文件
# 对于不同的项目或数据库,指定不同的配置文件,生成不同的json
def dynamic_import():
import importlib.util
argv = sys.argv
if len(argv) <= 1:
print("请输出加载的python配置模块名!")
exit(-1)
module_ = argv[1]
try:
print(f"使用__import__导入模块")
module = __import__(module_)
except Exception as e:
print(f"使用__import__导入模块失败")
print(f"使用importlib导入模块")
args = module_.split(os.sep)
if len(args) == 1:
module_name = args[0]
module_path = module_name
elif len(args) > 1:
module_name = args[-1]
module_path = module_
print(f"module_path:{module_path}\nmodule_name:{module_name}")
spec = importlib.util.spec_from_file_location(module_name, module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
m_keys = module.__dict__
key_list = list(m_keys.keys())
for k in key_list:
if not str(k).startswith("__"):
globals()[k] = m_keys.get(k)
return module
dynamic_import()
global_config = project_config
config_source = source_ds
project_path = global_config['project_path']
hive_host = global_config['hive_host']
hive_port = global_config['hive_port']
hive_db = global_config['hive_db']
use_kb = global_config['enable_kerberos']
use_pt = global_config['enable_partition']
keytab = global_config['key_table']
principal = global_config['principal']
# 加载当前项目数据库及表
def load_db():
if isinstance(config_source, list):
# 多数据源多数据库模式
for source in config_source:
db_tables_ = source["db_tables"]
db_connect = source["connect"]
host_ = db_connect['host']
port_ = db_connect['port']
username_ = db_connect['username']
password_ = db_connect['password']
for db_info in db_tables_:
db_ = db_info["db"]
if dict(db_info).keys().__contains__("project"):
project_ = db_info["project"]
else:
project_ = None
tables_ = db_info["tables"]
query_table(host_, port_, username_, password_, db_, project_, tables_)
else:
print("加载source_ds的config配置出现问题...")
def save_local(save_path, datax_json):
path = Path(f'../job/{save_path}')
if datax_home is not None:
path = Path(f"{datax_home}/job/{save_path}")
elif not Path('../').exists():
path = Path(f"job/{save_path}")
path.parent.mkdir(parents=True, exist_ok=True)
# 覆盖文件写入
path.write_text(datax_json, encoding="utf-8")
def camel_to_snake(field: str):
return re.sub('([a-z0-9])([A-Z])', r'\1_\2', field).lower()
def is_camel(s):
return bool(re.match(r'^[a-z]+([A-Z][a-z]*)*$', s))
def convert_field(field):
table_name = field[0]
field_name = field[1]
field_type = field[2]
field_comment = field[3]
# 是否为驼峰
if is_camel(field_name):
table_name = f"camel_{table_name}"
field_name = camel_to_snake(field_name)
field_comment = f"({field[1]}){field_comment}"
return [table_name, field_name, field_type, field_comment]
def convert_ods_field(field):
field_name = field['field_name']
field_type = field['field_type']
field_hive_type = field['field_hive_type']
field_comment = field['field_comment']
# 是否为驼峰
if is_camel(field_name):
field_name = camel_to_snake(field_name)
field_comment = f"({field['field_name']}){field_comment}"
return {"field_name": field_name, "field_type": field_type, "field_hive_type": field_hive_type,
"field_comment": field_comment}
def build_db(tables: list):
database = {}
for table in tables:
# 查询指定表的所有字段名和类型
table_name = table[0]
field_name = table[1]
field_type = table[2]
field_comment = table[3]
table_fields: list = database.get(table_name)
field_hive_type = hive_type(field_type)
field_one = {"field_name": field_name, "field_type": field_type, "field_hive_type": field_hive_type,
"field_comment": field_comment}
if table_fields is not None:
table_fields.append(field_one)
else:
table_fields = [field_one]
database[table_name] = table_fields
return database
def run_sub_shell(cmd):
try:
# cmd = f"source /etc/profile && cd {datax_home} && " + cmd
# print(cmd)
output = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT, encoding="utf-8", cwd=datax_home)
print(f'Command "{cmd}" executed successfully. Output: \n{output}')
return output
except subprocess.CalledProcessError as e:
print(f'Command "{cmd}" failed with error: \n{e.output}')
exit(-1)
def hive_file_sql(create_db_sql):
# 创建临时文件
tmp_hql = f"{datax_home}/tmp/_hive_sql.hql"
with open(tmp_hql, mode="w", encoding="utf-8") as fp:
fp.write(create_db_sql)
# 执行hive -f
if os.path.exists(tmp_hql):
run_sub_shell(f"hive -f {tmp_hql}")
else:
print(f"{tmp_hql}文件不存在...")
# 删除临时文件
os.remove(tmp_hql)
def query_table(host, port, user, password, db, project_, include_tables):
# 连接 MySQL 数据库
conn = mysql.connector.connect(
host=host,
port=port,
user=user,
password=password,
database=db
)
# 获取游标对象
cursor = conn.cursor()
query_col_sql = f"select table_name,column_name,data_type,column_comment from information_schema.`COLUMNS` where table_schema='{db}' "
if len(include_tables) > 0:
name_str = ",".join([f"'{x}'" for x in include_tables])
table_filter = f' and table_name in({name_str})'
query_col_sql += table_filter
else:
print(f"查询数据库:[{db}]的所有表")
# 查询指定数据库中的所有表名
# print(query_col_sql)
cursor.execute(query_col_sql)
tables = cursor.fetchall()
# 数据库的json
database = build_db(tables)
create_db_sql = f"use {hive_db};"
# 生成各个表的datax配置文件
for table_name in database.keys():
table_fields = database[table_name]
ods_source, ods_table, datax_json = build_datax(host, port, user, password, db, project_, table_name,
table_fields)
# datax和hive的表名全部小写,datax配置文件中的表名使用原始的大小写
save_local(f"{hive_db}/{ods_table}.json", datax_json)
print(f"生成datax配置文件-->{hive_db}/{ods_table}.json")
# 生成建表语句
create_db_sql += build_create_hive(ods_table, table_fields)
print(f"创建hive表-->{hive_db}.{ods_table}")
hive_file_sql(create_db_sql)
# print(create_db_sql)
# 关闭游标和连接
cursor.close()
conn.close()
print(f"自动处理数据库[{db}]的datax配置及hive库ods表完成\n")
def exec_hive_sql(sql_list=["show databases"]):
# 连接到Hive服务器
if use_kb:
conn = hive.Connection(host=hive_host, port=hive_port, database=hive_db, auth='KERBEROS',
kerberos_service_name='hive')
else:
conn = hive.Connection(host=hive_host, port=hive_port, database=hive_db)
# 执行查询
cursor = conn.cursor()
for sql in sql_list:
# print(f"执行sql:\n{sql}\n")
cursor.execute(sql)
# 关闭连接
cursor.close()
conn.close()
def build_create_hive(hive_table, fields):
# 生成建表语句
stored = "orc"
hive_fields = list(map(convert_ods_field, fields))
field_sql = ",\n".join(
map(lambda x: f"\t\t`{x['field_name']}` {x['field_hive_type']} comment '{x['field_comment']}'", hive_fields))
dw_type = hive_table.split("_")[0]
partition_sql = ""
if use_pt:
partition_sql = "partitioned by(pt_day string comment '格式：YYYYMMDD')"
create_sql = f"""
drop table if exists {hive_table};
create external table if not exists {hive_table}
(
{field_sql}
){partition_sql}
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
stored as {stored}
LOCATION '{project_path}/{dw_type}/{hive_table}'
TBLPROPERTIES('orc.compress'='SNAPPY');
"""
# print(create_sql)
return create_sql
def unify_name(database: str):
snake_case = re.sub(r'(?<!^)(?=[A-Z])', '_', database).lower()
return snake_case.replace("_", "").replace("-", "")
def build_datax(host, port, username, password, database, project_, source_table, fields):
if project_ is None or len(project_) == 0:
ods_source = unify_name(database)
else:
ods_source = unify_name(project_)
pt_str = '_pt' if use_pt else ''
ods_table = f"ods_{ods_source}_{source_table}{pt_str}_a".lower()
jdbc_url = f"jdbc:mysql://{host}:{port}/{database}?useSSL=false&useUnicode=true&allowMultiQueries=true&characterEncoding=utf8&characterSetResults=utf8&serverTimezone=Asia/Shanghai"
columns = ",".join([f'"`{field["field_name"]}`"' for field in fields])
hive_fields = list(map(convert_ods_field, fields))
hive_columns = ",".join(
[f'{{"name":"{field["field_name"]}","type":"{field["field_hive_type"]}" }}' for field in hive_fields])
pt_config = '"ptDay": "pt_day",' if use_pt else ""
kerberos_config = f'"haveKerberos": "{use_kb}","kerberosKeytabFilePath": "{keytab}","kerberosPrincipal": "{principal}",' if use_kb else ""
mysql_hive_tpl = '''
{
"job": {
"setting": {
"speed": {
"channel": 3,
"byte":-1
},
"errorLimit": {
"record": 0,
"percentage": 0
}
},
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "${username}",
"password": "${password}",
"column": [${columns}],
"splitPk": null,
"connection": [
{
"table": ["${sourceTable}"],
"jdbcUrl": ["${jdbcUrl}"]
}
]
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"defaultFS": "hdfs://master:8020",
"fileType": "orc",${kerberosConfig}${ptConfig}
"path": "${projectPath}/ods/${odsTable}",
"fileName": "${odsTable}",
"column": [${hiveColumns}],
"writeMode": "truncate",
"fieldDelimiter": "\\t",
"compress": "SNAPPY",
"database":"${database}"
}
}
}
]
}
}
'''
var_dict = {"username": username, "password": password, "columns": columns, "sourceTable": source_table,
"jdbcUrl": jdbc_url,
"kerberosConfig": kerberos_config, "ptConfig": pt_config, "projectPath": project_path,
"odsTable": ods_table,
"hiveColumns": hive_columns, "database": hive_db}
for k in var_dict.keys():
mysql_hive_tpl = mysql_hive_tpl.replace('${' + k + '}', var_dict[k])
data = json.loads(mysql_hive_tpl)
data = json.dumps(data, indent=2, ensure_ascii=False).replace("True", "true").replace("False", "false")
return ods_source, ods_table, data
def hive_type(mysql_type):
if mysql_type == "tinyint" or mysql_type == "smallint" or mysql_type == "boolean":
return "smallint"
elif mysql_type == "long" or mysql_type == "int":
return "bigint"
elif mysql_type == "float" or mysql_type == "double" or mysql_type == "decimal":
return "double"
elif mysql_type == "date":
return "date"
elif mysql_type == "timestamp":
return "timestamp"
else:
return "string"
if __name__ == '__main__':
load_db()配置文件模版config_xx.py
# 目前只支持从mysql到hive
project_name = "project_name"
hive_host = "master"
hive_db = "hive_db"
# hdfs路径
if project_name == 'project_name':
project_path = f"/project/{project_name}/{hive_db}"
else:
project_path = f"/project/{project_name}/warehouse/{hive_db}"
# 主要配置
project_config = {
"project_name": project_name,
"hive_host": hive_host,
"hive_port": 10000,
"project_path": project_path,
# write:hive数据库
"hive_db": hive_db,
"enable_kerberos": True, #是否启用Kerberos
"enable_partition": False, #是否分区表
# 默认不用修改
"default_fs": f"hdfs://{hive_host}:8020",
"principal": "principal",
"key_table": "hdfs.keytab" #如果存在Kerberos的话
}
# 不指定表名,则扫描全库
# 抽取的mysql数据源
source_ds = [
{
"db_tables": [
{
"db": "database",
"tables": ["view_online"],
}
],
"connect": {
"host": "host",
"port": 23306,
"username": "username",
"password": "password",
}
}
]使用方法:
配置DATAX_HOME,将脚本放在$DATAX_HOME/bin,自行创建job文件夹
生成Datax任务json,同时生成的json文件在$DATAX_HOME/job/{hive_database}下面
python3 build_core.py config/config_dw.py多线程执行
cd $DATAX_HOME
python3 bin/datax_run.py --hive hive_database--hive 指定hive数据库,指定要执行的json文件路径在$DATAX_HOME/job/{hive_database}
