poyee-data-warehouse/bin/datax-hive-import-starter.py

#!/usr/bin/env /usr/bin/python3
# -*- coding:utf-8 -*-
"""
DataX hive-import 入口：目标=Hive（自动预建分区），对应 jobs/raw/ 场景。

老参数平迁集（见 kb/90 §2.6）：
  -ini <file>          单 ini，可多次
  -inis <dir>          ini 目录（非递归扫 *.ini），可多次
  -start-date / -stop-date  yyyyMMdd
  -host <hostname>     显式指定 worker
  -random              加权随机选 worker
  -parallel            并行（默认串行）
  -skip-datax          只生成 json 不执行
  -skip-partition      跳过 Hive 分区预建
  -t <db.table>        显式追加需建分区的表，可多次
"""
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()