#!/usr/bin/env /usr/bin/python3 # -*- coding:utf-8 -*- """ DataX hive-import 入口:目标=Hive(自动预建分区),对应 jobs/raw/ 场景。 老参数平迁集(见 kb/90 §2.6): -ini 单 ini,可多次 -inis ini 目录(非递归扫 *.ini),可多次 -start-date / -stop-date yyyyMMdd -host 显式指定 worker -random 加权随机选 worker -parallel 并行(默认串行) -skip-datax 只生成 json 不执行 -skip-partition 跳过 Hive 分区预建 -t 显式追加需建分区的表,可多次 """ import argparse import os import sys from datetime import date, timedelta project_root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(project_root_dir) from dw_base.datax.entry import DataxImport def main(): parser = argparse.ArgumentParser( prog='datax-hive-import-starter', description='DataX hive-import 入口:PG/其他源 → Hive (raw 层)', ) parser.add_argument('-ini', action='append', default=[], metavar='FILE', help='DataX ini 单文件(可多次)') parser.add_argument('-inis', action='append', default=[], metavar='DIR', help='DataX ini 目录,非递归扫 *.ini(可多次)') parser.add_argument('-start-date', default=None, metavar='YYYYMMDD', help='默认昨天(对齐老 sh 行为)') parser.add_argument('-stop-date', default=None, metavar='YYYYMMDD', help='默认今天(对齐老 sh 行为)') parser.add_argument('-host', default=None, metavar='HOSTNAME', help='显式指定 worker(优先于 -random)') parser.add_argument('-random', action='store_true', dest='use_random', help='从 conf/workers.ini 加权随机选 worker') parser.add_argument('-parallel', action='store_true', help='并行执行(默认串行)') parser.add_argument('-skip-datax', action='store_true', help='只生成 json 不执行 datax.py') parser.add_argument('-skip-check', action='store_true', help='跳过 HDFS 源路径 check(默认开启 check,missing/empty 算失败;显式关闭后不检查直接交 datax)') parser.add_argument('-skip-partition', action='store_true', help='跳过 Hive 分区预建') parser.add_argument('-t', action='append', default=[], dest='extra_partition_tables', metavar='DB.TABLE', help='显式追加需建分区的 Hive 表(可多次)') args = parser.parse_args() # 默认日期:昨天 → 今天(对齐老 datax-single-job-starter.sh:207-219 行为) if not args.start_date: args.start_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d') if not args.stop_date: args.stop_date = date.today().strftime('%Y%m%d') print('{script} 收到参数: {argv}'.format( script=os.path.basename(__file__), argv=' '.join(sys.argv[1:]), )) print(' start_date={s} stop_date={e}'.format(s=args.start_date, e=args.stop_date)) importer = DataxImport( base_dir=project_root_dir, workers_ini_path=os.path.join(project_root_dir, 'conf', 'workers.ini'), release_user=os.environ['RELEASE_USER'], release_root_dir=os.environ['RELEASE_ROOT_DIR'], python3_path=os.environ['PYTHON3_PATH'], datax_home=os.environ['DATAX_HOME'], log_root_dir=os.environ['LOG_ROOT_DIR'], ) failed = importer.run( inis=args.ini, inis_dirs=args.inis, start_date=args.start_date, stop_date=args.stop_date, host=args.host, use_random=args.use_random, parallel=args.parallel, skip_partition=args.skip_partition, skip_datax=args.skip_datax, skip_check=args.skip_check, extra_partition_tables=args.extra_partition_tables, ) sys.exit(failed) if __name__ == '__main__': main()