poyee-data-warehouse/jobs/raw/usr/raw_usr_traces_apd_d.py

#!/usr/bin/env /usr/bin/python3
# -*- coding:utf-8 -*-
"""
埋点 gz 入仓 raw 包装脚本。

逐日：本地 gz（固定服务器 /data/upload/traces/traces-YYYY-MM-DD.json.gz）
  → hdfs put 到临时目录 /tmp/raw_usr_traces/{dt}/
  → 调 bin/spark-sql-starter.py 跑解析脱敏 SQL（写 raw 当日分区）
  → 清临时目录

CLI：-dt 支持单日 / `20260601-` / 区间 / 离散（复用 get_date_range）。
缺文件跳过（缺数据容忍）。返回非 0 = 失败的日期数。
"""
import os
import subprocess
import sys

PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..', '..'))
sys.path.insert(0, PROJECT_ROOT)

from dw_base.common.config_constants import K_DT
from dw_base.utils.config_utils import parse_args
from dw_base.utils.datetime_utils import get_date_range, get_yesterday

LOCAL_DIR = '/data/upload/traces'
HDFS_TMP_BASE = '/tmp/raw_usr_traces'
SQL = 'jobs/raw/usr/raw_usr_traces_apd_d.sql'
UDF = 'dw_base/udf/business/spark_traces_udf.py'
STARTER = 'bin/spark-sql-starter.py'


def run(cmd, cwd=None):
    print('+ ' + ' '.join(cmd))
    return subprocess.call(cmd, cwd=cwd)


def main():
    config, _ = parse_args(sys.argv[1:])
    date_range = get_date_range(config.get(K_DT, get_yesterday()))
    failed = []
    for dt in date_range:  # dt = yyyymmdd
        local_gz = '%s/traces-%s-%s-%s.json.gz' % (LOCAL_DIR, dt[0:4], dt[4:6], dt[6:8])
        if not os.path.isfile(local_gz):
            print('跳过 %s：本地 gz 不存在 %s' % (dt, local_gz))
            continue
        hdfs_tmp = '%s/%s' % (HDFS_TMP_BASE, dt)
        run(['hdfs', 'dfs', '-rm', '-r', '-f', hdfs_tmp])
        run(['hdfs', 'dfs', '-mkdir', '-p', hdfs_tmp])
        if run(['hdfs', 'dfs', '-put', '-f', local_gz, hdfs_tmp]) != 0:
            print('!! %s hdfs put 失败' % dt)
            failed.append(dt)
            continue
        rc = run(['python3', STARTER, '-f', SQL, '-u', UDF, '-dt', dt], cwd=PROJECT_ROOT)
        run(['hdfs', 'dfs', '-rm', '-r', '-f', hdfs_tmp])
        if rc != 0:
            print('!! %s spark-sql 失败 (rc=%d)' % (dt, rc))
            failed.append(dt)
    if failed:
        print('失败日期：%s' % ','.join(failed))
    sys.exit(len(failed))


if __name__ == '__main__':
    main()