90-重构路线.md 26 KB

重构路线

基于老项目 tendata-warehouse-release 的代码分析,为新项目 poyee-data-warehouse 规划的重构路线。 本文档说"为什么改、怎么改";配套的 92-重构进度.md 说"改到哪一步了"。

一、模块重命名(高优先级)

1.1 tendatadw_base

影响范围:

类型 涉及内容 处理方式
Python import from tendata import *from tendata.xxx import yyy 全局替换
SQL 中的 ADD FILE ADD FILE tendata/spark/udf/xxx.py 全局替换
zip 打包命令 zip -qr tendata.zip tendata 改为 dw_base.zip dw_base
addPyFile sparkContext.addPyFile('tendata.zip') 改为 dw_base.zip
路径匹配正则 re.sub(r"tendata-warehouse.*", ...) 更新为新项目名
目录名引用 PROJECT_ROOT_PATH 相关逻辑 自动适配

建议做法: 先用脚本批量替换,再逐文件审查。注意 tendata 字样可能出现在老项目的 Hive 数据库名、表名或 DataX 数据源 ini 名中,这些属于老业务数据、不应替换——新项目业务 SQL 从零开发,不涉及老库老表。

1.2 launch-pad/jobs/(不做业务迁移,仅建立新结构)

重要澄清:老项目 launch-pad/ 中的业务代码与新项目业务完全无关(属于上个项目的历史业务),不存在内容迁移launch-pad/ 在重构期间作为样板 SQL 代码的参考(看写法、看 ${dt} 用法、看 DataX ini 结构),新项目所有业务 SQL 从零开发完成后,launch-pad/ 整体删除。

jobs/ 目录按数仓分层 × 业务域二维组织,结构见 00-项目架构.md §5 的样板。业务域代码统一使用命名规范定义的 trd/usr/prd/shp/pub/dim(见 21-命名规范.md §3.2)。

需要处理的代码引用

  • 脚本中 list_files('launch-pad/...') 的硬编码路径 → 改为 jobs/...
  • zip -qr launch-pad.zip launch-pad 类命令(如有)→ 改为 jobs
  • DolphinScheduler 工作流里老路径的引用 → 新项目上线时一并替换

二、消除硬编码(高优先级)

2.1 当前硬编码清单

硬编码内容 所在位置 建议方案
DATAX_HOME=/opt/module/datax bin/common/init.sh 移入 conf/env.sh 或环境变量
PYTHON3_PATH="/usr/bin/python3" bin/common/init.sh 移入 conf/env.sh
RELEASE_USER="alvis" bin/common/init.sh 改为 RELEASE_USER="bigdata" 并移入 conf/env.sh
RELEASE_ROOT_DIR="/home/alvis/release" init.sh__init__.py 改为 /home/bigdata/release 并移入 conf/env.sh
项目部署目录 poyee-data-warehouse/ publish.sh 新项目发布目录为 /home/bigdata/release/poyee-data-warehouse/
DATAX_WORKERS=(m3 d1 d2 d3 d4) + 权重 init.sh 移入 conf/workers.conf
HADOOP_CONF_DIR='/etc/hadoop/conf' __init__.py 使用系统环境变量
LOG_ROOT_DIR="/opt/data/log" init.sh__init__.py 移入 conf/env.sh
钉钉 access_token dingtalk_notifier.py 移入 conf/alerter.conf(敏感项)
企微 Webhook Key dw_base/common/alerter_constants.py 外移到 conf/alerter.ini入库——部署靠 git pull,gitignore 会拉不到;webhook key 不算高敏感,最多被拿去发垃圾消息),Python 侧改 ConfigParser 加载;alerter_constants.py 整个删除
DS API 地址 ds/config/base_config.yaml 已在 yaml,保持即可
Spark 默认参数(executor/driver/shuffle/sql.*) dw_base/spark/spark_sql.py 构造函数 + .config(...) 移入 conf/spark-defaults.yaml,SQL 文件可用 SET 覆盖,见 §2.3
DataX ini 路径前缀剥离 conf/datax/config/ bin/datax-single-job-starter.sh(TEMP 处理)、bin/datax-job-config-generator.pyreplace('conf/datax/config/', ''))、bin/datax-multiple-job-starter.sh(日志路径派生) 原目录已整体挪到 conf/bak/ 并 gitignore,脚本里 replace 现在是 no-op 死逻辑。去除前缀假设,改为靠 ini 文件名(= 任务唯一标识,见 21-命名规范.md §3.9)识别用途
DataX 生成 JSON 输出目录名 conf/datax/generated bin/datax-job-config-generator.py 末尾 default_output_dirbin/datax-single-job-starter.sh 第 89/118 行、bin/datax-multiple-job-starter.sh 第 187 行、.gitignore 目录改名 conf/datax-json/;子路径扁平化为 conf/datax-json/{env}/{ini_basename}.json(仅按 env 分一级,去掉 src_dst / project_layer_env 等派生层级);.gitignore 同步改
JOB_NAME / JSON 文件名的 ini→json 转换逻辑重复实现 Python 侧 bin/datax-job-config-generator.py:126os.path.basename(gcf).replace('.ini', '.json'))+ Bash 侧 bin/datax-single-job-starter.sh:88basename .ini 合一到 dw_base.datax.path_utils.job_name_from_ini()(或类似工具);Bash 侧通过 python3 -c 调用或在 bin/common/init.sh 定义等价 shell 函数,单一来源
DataX 脚本不支持 -env 参数 bin/datax-*.sh / bin/datax-job-config-generator.py 参数解析 新增 -env 参数,由 ini 内 dataSource = {db_type}/{instance} 拼接成 datasource/{db_type}/{env}/{instance}.ini 的完整路径;详见 §2.5
datasource/ 单层组织(无环境子目录) datasource/{db_type}/{instance}.ini 改为 datasource/{db_type}/{env}/{instance}.ini(运维侧权限隔离 + 支持一套代码跑多环境)
ini 里 dataSource 字段拼接环境后缀 老项目写法 dataSource = pg-hobby-prod 改为 dataSource = {db_type}/{instance}(不含环境),env 由脚本注入
导出类 ini 扇出撞名风险 jobs/ads/{域}/ 下 ini 若都以源 Hive 表名命名,同一张 ads 表扇出到多个目标库时会重名覆盖 命名规则改为 {源 Hive 表名}__{目标 db_type}_{目标 instance}.ini(双下划线分隔源/目标),见 21-命名规范.md §3.9
dw_base/common/template_constants.py 大量死代码 定义了 20+ 个 SQL 模板路径常量,实际只有 2 个(MYSQL_HIVE_CREATE_TABLE_TEMPLATE / MYSQL_HIVE_HBASE_CREATE_TABLE_TEMPLATE)被引用,其余 18 个零 import 整个文件删除;连带废弃下一条
MySQLReader.generate_hive_ddl() / generate_hive_over_hbase_ddl() 自动建表 DDL 路径 dw_base/datax/plugins/reader/mysql_reader.py:195/222,被 bin/datax-gc-generator.py:616/728 调用;且 conf/template/ 目录在新项目根本不存在,真调用会 FileNotFoundError 整段路径废弃——与 CLAUDE.md 约定的 manual/ddl/ 是 DDL 唯一来源相冲突。datax-gc-generator.py 仅生成 ini 配置,不再输出 CREATE TABLE DDL;DDL 由开发者按 21-命名规范.md 手写到 manual/ddl/
缺少集中的人/AI 参考模板目录 —(新增) 已建 conf/templates/{datasource,datax/{raw,ads,manual},sql,ddl}/,模板用 *.template.{ini,sql} 双扩展名。与上条废弃的运行时模板完全不同:这里的模板不被任何代码读取,只供人/AI 对照写新文件;kb/README.md 已加入口

2.2 建议的配置结构

conf/
├── env.sh                    # Shell 环境变量(路径、用户、日志目录等)
├── env.py                    # Python 环境变量(或直接读 env.sh)
├── workers.conf              # DataX Worker 列表与权重
├── alerter.conf              # 告警 Webhook 配置(敏感项,可 .gitignore)
├── spark-defaults.yaml       # Spark 默认参数
└── ds/
    ├── base_config.yaml      # DolphinScheduler 配置
    └── process_code.yaml     # 工作流编码映射

2.3 Spark 配置三级覆盖策略

现状dw_base/spark/spark_sql.py 构造函数里硬编码了约 15 个 .config(...) 调用(executor/driver/memory/parallelism/shuffle/adaptive/arrow/codegen 等),默认值写死在构造参数里,覆盖只能通过 SparkSQL 构造函数传参或 SQL 文件内 SET

问题

  • 想批量调整 shuffle partitions 的默认值,就得改代码 + 发版
  • 不同类型的作业(dwd 大宽表 / ads 小聚合)需要不同默认,现状只能每张表的 SQL 开头都重复写一遍 SET
  • 默认参数和业务代码耦合,不便于运维按集群负载动态调整

目标态:三级覆盖

conf/spark-defaults.yaml         (L1) 全局默认,运维可改,发版同步到集群
        ↓ 被覆盖
SQL 文件内 SET spark.xxx=yyy     (L2) 单作业级别的覆盖,业务开发写
        ↓ 被覆盖
命令行 -sc key=value / Python 构造函数传参  (L3) 临时/调试 override

conf/spark-defaults.yaml 草案

# 全局 Spark 默认参数,dw_base/spark/spark_sql.py 启动时加载
# 单作业需要覆盖时,在对应 jobs/*.sql 文件开头写 SET;不要改本文件

executor:
  instances: 4
  cores: 4
  memory: 8g
  memoryOverhead: 2g

driver:
  cores: 2
  memory: 4g
  maxResultSize: 2g

sql:
  shuffle.partitions: 200
  adaptive.enabled: true
  broadcastTimeout: -1
  codegen.wholeStage: false
  execution.arrow.enabled: true
  execution.arrow.fallback.enabled: true
  files.ignoreCorruptFiles: true
  statistics.fallBackToHdfs: true

default:
  parallelism: 400

代码改动要点

  1. dw_base/spark/spark_sql.py
    • 新增 _load_default_config() -> dict:读 conf/spark-defaults.yaml,扁平化为 {"spark.executor.instances": 4, ...} 形式(dot-notation 按 yaml 嵌套路径拼)
    • 构造函数接收的显式参数(spark_executor_cores 等)改为 None 默认,若未传则 fall back 到 yaml
    • SparkSession.builder.config(...) 链改成 for k, v in resolved_config.items(): builder.config(k, v)
  2. SQL 文件内的 SET spark.xxx=yyy 本来就由 spark.sql(...) 原生支持,无需改动
  3. 命令行 -sc 参数保持现有语义,覆盖 L1
  4. Python 单测要能跑:yaml 读取要容错(测试环境下找不到 conf 文件时回退到一套最小内置默认,不阻塞 tests/unit/

兼容性:老代码里已在写 SparkSQL(spark_executor_cores=8, ...) 的调用站点不破坏,因为显式传参仍是最高级(L3)。

落地时的两个坑

  1. L2 覆盖只对 spark.sql.* 系参数生效。Spark 的参数分两类:
    • spark.sql.*spark.default.parallelism 等运行时参数 —— spark.conf.set(...) 或 SQL 内 SET 可动态改写
    • spark.executor.*spark.driver.*spark.executor.memoryOverhead 等资源类参数 —— session 启动时锁定,SQL 里写 SET spark.executor.memory=16g 不会真的扩容已启动的 executor

因此开发写 SQL 内 SET 时只能调 spark.sql.* 和并行度;需要改资源的场景只能走 L3(命令行 -sc 或调用方在构造 SparkSQL(...) 时显式传参)。文档里和 spark-defaults.yaml 注释里都要讲清楚这条,避免开发以为 SET spark.executor.memory 有效。

  1. conf/spark-defaults.yaml 的路径解析依赖 PROJECT_ROOT_PATH,这和 §三 __init__.py 瘦身存在先后依赖:
    • 现状 PROJECT_ROOT_PATHdw_base/__init__.py 顶部定义,import dw_base 就会拿到
    • 瘦身后 __init__.py 只保留最基本路径定义,PROJECT_ROOT_PATH 仍可用,但拆分过程中要保证 spark_sql.py 加载 yaml 的那行代码拿到的根路径与瘦身前一致
    • 执行顺序建议:先做 §三 __init__.py 瘦身,把 PROJECT_ROOT_PATH 的定义稳定下来;再做 §2.3 的 spark-defaults.yaml 接入。反过来做会踩到"瘦身后路径变了"的返工

2.4 项目根 .gitignore

现状:老项目根目录没有 .gitignore.idea/workspace.xml.claude/settings.local.json 等个人状态文件随时可能被误提交,conf/alerter.conf(规划中的告警 Webhook,见 §2.2)也需要挡在版本控制外。

目标:在项目根新建 .gitignore,在阶段 2 建立 conf/ 目录的同一节奏下一起落地(顺序上先有 .gitignore 再把 alerter.conf 放进 conf/,避免敏感文件误入第一次提交)。

内容清单

# ---- Claude Code 本地设置 ----
.claude/settings.local.json

# ---- JetBrains 个人工作区 ----
# 注意:.idea/ 不整体 ignore —— modules.xml / *.iml / inspectionProfiles/ 是
# 团队可共享的项目结构配置,保留入库对新成员友好(开箱即用),这也是
# JetBrains 官方推荐做法
.idea/workspace.xml
.idea/tasks.xml
.idea/shelf/
.idea/usage.statistics.xml
.idea/dictionaries/
.idea/httpRequests/

# ---- Python / 构建产物 ----
__pycache__/
*.py[cod]
*.egg-info/
.pytest_cache/
.venv/
venv/

# ---- 运行期产物 ----
*.log
dw_base.zip

# ---- 敏感配置(运行时自动从 datasource/ 注入或在 conf/ 本地覆盖) ----
conf/alerter.conf

注意事项

  1. .idea/ 不整体 ignore
    • 入库:modules.xml*.imlinspectionProfiles/(项目结构 + 代码检查规则,团队共享)
    • 忽略:workspace.xmltasks.xmlshelf/usage.statistics.xml 等个人/统计文件
  2. .claude/ 也不整体 ignoresettings.jsoncommands/agents/ 是团队共享配置;只忽略 settings.local.json
  3. dw_base.zipspark_sql.py 运行时生成的 PySpark 打包产物,属于构建产物不入库
  4. conf/alerter.conf 一开始就放进 .gitignore:阶段 2 迁移钉钉/企微 Webhook 时,新建文件前 .gitignore 必须先就位

与仓库改名的联动

仓库改名 tendata-warehouse-releasepoyee-data-warehouse 时(阶段 1 尾声),.idea/tendata-warehouse-release.iml 也要改名为 .idea/poyee-data-warehouse.iml,并同步更新 .idea/modules.xml 里的引用。这一步不属于 .gitignore 的范畴,但和它是同一天会碰到的事,在阶段 1 的仓库改名 checklist 里一起记一笔。

2.5 DataX 脚本多环境支持与路径解耦

现状(脚本残留老逻辑,新项目的业务 ini 未走这条路径):

  1. 脚本里残留路径前缀剥离bin/datax-single-job-starter.shbin/datax-job-config-generator.py 仍通过剥离 conf/datax/config/ 前缀从源 ini 路径里派生 SRC_DST / PROJECT_LAYER_ENV,用于生成 JSON 输出路径和 datax-multiple-job-starter.sh 的日志目录。该目录已整体挪到 conf/bak/ 并 gitignore,新项目 ini 放在 jobs/raw/{domain}/ / jobs/ads/{domain}/ / manual/,前缀不匹配时剥离变成 no-op,输出会落到形如 conf/datax/generated/jobs/raw/trd/xxx.json 的位置——能跑但与新约定不符。代码里这段死逻辑要清理,同时输出根目录也一并改名为 conf/datax-json/
  2. 没有 -env 参数:所有脚本不认 -env
  3. 数据源配置单层组织:老约定 datasource/{db_type}/{instance}.ini 把环境和实例扁平混在一起,要么靠不同的 {instance} 名字(如 hobby-prod / hobby-dev)区分,要么靠部署时替换文件。第一种让 ini 里写 dataSource = pg-hobby-prod 这种绑死环境的字符串;第二种让开发侧不知道当前跑的是哪套。

目标态:一套代码多环境运行

三件事联动(顺序重要):

阶段 1:datasource 改按环境分子目录

  • 改为 datasource/{db_type}/{env}/{instance}.ini
  • 运维在集群侧按 prod / test / dev 建子目录,各自放一套连接配置
  • 权限隔离更方便:prod 子目录只给生产账号可读

阶段 2:ini 里 dataSource 字段去掉环境后缀

  • 老写法:dataSource = pg-hobby-prod
  • 新写法:dataSource = pg/hobby(只写 {db_type}/{instance},不含 env)
  • 变更项目下所有存量 ini:当前 conf/bak/ 下的老配置不处理,等业务新 ini 从零写起时就按新规范

阶段 3:DataX 脚本加 -env 参数 + env 注入逻辑

bin/common/init.sh 扩展:

# 从命令行参数里挑出 -env,或 fall back 到 conf/env.sh
# 不读环境变量(用户明确要求不污染 shell 环境)
# 不做"按用户/部署目录自动派生"——env 必须来自 -env 或 conf/env.sh,二者都没给就退出
resolve_env() {
  for arg in "$@"; do
    case $arg in
      -env) NEXT_IS_ENV=1 ;;
      -env=*) DW_ENV="${arg#*=}" ;;
      *) [ -n "$NEXT_IS_ENV" ] && DW_ENV="$arg" && unset NEXT_IS_ENV ;;
    esac
  done
  if [ -z "$DW_ENV" ]; then
    # fall back 到 conf/env.sh(入仓库的默认配置)
    [ -f "${BASE_DIR}/conf/env.sh" ] && . "${BASE_DIR}/conf/env.sh"
  fi
  if [ -z "$DW_ENV" ]; then
    echo "[FATAL] DW_ENV not set: pass -env <dev|test|prod> or define DW_ENV in conf/env.sh" >&2
    exit 1
  fi
  export DW_ENV
}

conf/env.sh 草案入仓库,开发者维护,服务本地调试):

# 全局环境变量默认值
# env 判定优先级:命令行 -env > 本文件 DW_ENV
# 默认锁定为 dev:本地调试开箱即用;DolphinScheduler / 生产脚本总是命令行显式 -env prod 覆盖
DW_ENV=dev
# LOG_ROOT_DIR=/opt/data/log

dw_base/datax/job_config_generator.py 改造

  • JobConfigGenerator.__init__ 接受新参数 env
  • 各 reader/writer 插件在解析 dataSource = pg/hobby 时,自动拼成 datasource/pg/{env}/hobby.ini 的完整路径再去读连接信息

bin/datax-job-config-generator.py 改造

  • 新增 -env 参数
  • 去掉 temp = os.path.dirname(gcf).replace('conf/datax/config/', '').split('/') 这段路径前缀剥离
  • JSON 输出目录从 conf/datax/generated/ 改名为 conf/datax-json/,子路径简化为 conf/datax-json/{env}/{ini_basename}.json(扁平,按 env 分一级;ini_basename21-命名规范.md §3.9 保证全局唯一——采集类 = 目标 Hive 表名、导出类 = {源}__{db_type}_{instance}、一次性 = 日期前缀)
  • .gitignore 同步把 conf/datax/generated 改成 conf/datax-json

阶段 4:启动脚本层串起来

  • datax-single-job-starter.sh 调用 generator 时把 $DW_ENV 透传过去
  • datax-multiple-job-starter.sh 日志目录改为 ${LOG_ROOT_DIR}/datax/${DW_ENV}/${START_DATE}/${JOB_NAME}.log(把 SRC_DST / PROJECT_LAYER_ENV 彻底移除)

兼容性:阶段 1-3 期间 conf/bak/ 下的老 ini 不参与新流程,保留作为老项目历史资产;新业务 ini 全部从零按新规范写。

参考:kb/00-项目架构.md §4.3(DataX 脚本详细使用)、§6.4(多环境机制)、kb/21-命名规范.md §3.9(DataX ini 文件命名)。

三、__init__.py 瘦身(高优先级)

现状: tendata/__init__.py 约 120 行,import 即执行以下操作:

  • 环境变量设置
  • 颜色常量定义(30+ 个)
  • findspark.init()
  • 用户/权限/路径检测 + 打印
  • cow_says() 调用 shell

问题:

  • 任何 from dw_base import xxx 都会触发全部初始化
  • 不在 Spark 节点上运行的脚本也被迫执行 findspark.init()
  • 影响单元测试(测试 UDF 函数也要初始化 Spark 环境)

建议拆分为:

# dw_base/__init__.py —— 仅做最基本的路径定义
PROJECT_ROOT_PATH = ...
PROJECT_NAME = ...

# dw_base/core/env.py —— 环境检测(延迟调用)
# dw_base/core/colors.py —— 颜色常量
# dw_base/core/spark_env.py —— findspark 初始化(按需 import)

四、代码风格修正(中优先级)

4.1 __contains__ 反模式

全项目大量使用:

if config.__contains__(key):       # 反模式
if self.REGISTERED_UDF.__contains__(name):

应改为:

if key in config:                  # Pythonic
if name in self.REGISTERED_UDF:

4.2 Shell / Python 重复逻辑

bin/common/init.shdw_base/__init__.py 有大量重复的环境检测逻辑(用户判断、路径判断、日志目录、颜色常量)。

建议: 统一由 Python 入口处理,Shell 脚本仅做最小化的环境设置后调用 Python。或提取为一份共享的配置文件。

4.3 SQL 注入风险

mysql_utils.py 中使用 f-string 拼接 SQL:

sql = "... WHERE TABLE_SCHEMA='%s' ..." % (database, table_name)

建议: 改用参数化查询。

五、清理废弃代码(中优先级)

模块/文件 状态 建议
dw_base/validation/__init__.py 空文件 删除或实现数据质量校验
dw_base/ml/a.py 空文件 删除
dw_base/flink/__init__.py 空文件 删除(除非计划使用 Flink)
dw_base/elasticsearch/__init__.py 空文件 删除
dw_base/oss/oss2_util.py 使用场景不明 确认后决定保留或删除
dw_base/database/mongodb_utils.py 约 80% 是注释掉的旧代码 清理注释
conf/datax/ 下全部内容 已废弃的旧配置 保留少量样例,其余删除
sql_style.xml IntelliJ SQL 格式化规则 移入 .idea/ 或删除

六、测试体系搭建(中优先级)

6.1 现状

  • 仅 UDF 有少量 pytest 测试
  • 核心模块(SparkSQL、DataX 配置生成)无测试
  • 无 CI/CD 集成

6.2 建议的测试结构

tests/
├── conftest.py                    # pytest 公共 fixtures
├── unit/
│   ├── test_udf_trd.py            # UDF 单测(按业务域组织,纯函数,不依赖 Spark)
│   ├── test_udf_usr.py
│   ├── test_udf_pub.py
│   ├── test_config_utils.py       # 工具函数单测
│   ├── test_datetime_utils.py
│   ├── test_sql_utils.py
│   └── test_datax_generator.py    # DataX ini→json 生成测试
├── integration/
│   ├── test_spark_sql.py          # SparkSession local[*] 模式集成测试
│   └── test_hive_utils.py
└── quality/
    └── test_data_quality.py       # 数据质量校验(行数、空值率、主键唯一性)

6.3 测试策略

  • UDF 单测:纯 Python 函数,直接 assert,不需要 Spark 环境
  • DataX 配置生成测试:给定 ini 文件,断言生成的 JSON 结构正确
  • Spark 集成测试:使用 local[*] + 内存 Hive(enableHiveSupport() 需要 Hive MetaStore,可用嵌入式 Derby)
  • 数据质量:在 DolphinScheduler 工作流中加入校验节点

七、其他建议

7.1 依赖管理(已精简)

状态:2026-04-15 已完成首轮审计与精简。老清单 48 行 → 新清单 10 个强依赖,详见根目录 requirements.txt;原始快照备份在 requirements.txt.bak,并在注释里给每一行打了 [KEEP/DROP/LAZY/STDLIB] 结论。

精简策略

分类 处理方式 代表包
强依赖(KEEP) 留在 requirements.txt pyspark / pandas / pymongo / PyMySQL / requests / PyYAML / findspark / python-dateutil / wheel / pytest
无引用(DROP) 直接移除 openvino / transformers / scikit-learn / scipy / numpy / Flask / matplotlib / lxml / SQLAlchemy / jieba / cpca / openpyxl / xlrd / 等 20+ 个
stdlib 冗余(STDLIB) 移除 backport configparser —— Python 3 标准库自带,backport 安装反而会覆盖 stdlib
弱依赖(LAZY) 不写进 requirements,用到时手动 pip install elasticsearch / pyhive / redis / cryptography / oss2 / fuzzywuzzy / pygeohash / pypinyin —— 都只被即将清理的老业务代码引用

后续事项

  • LAZY 类依赖关联的老代码(tendata/scheduler/get_oldmongo_*mg2es/ent_interface_dingtalk*customs/similarity.pytendata/oss/oss2_util.pytendata/utils/excel_to_hive_utils.py)在阶段 4 / 阶段 5 清理废弃代码时一并删除,删完后即可彻底告别这些弱依赖
  • 不需要 requirements-base.txt / requirements-dev.txt 分文件——当前依赖规模下单文件已经足够
  • pyspark 2.4.0 暂保留(CDH 集群一致),等集群升级再一并上调

7.2 日志改进

  • pretty_print() 混合了控制台输出和文件写入,职责不清
  • Logging 类定义了但很少使用
  • 建议:统一使用 Python logging 模块,配置 handler 实现控制台+文件双输出

7.2.1 日志路径按 whoami 分流的硬编码逻辑

现状: bin/common/init.shdw_base/__init__.py 硬编码 RELEASE_USER="alvis",并按 whoami 是否等于该用户分流日志目录:

if [ "$(whoami)" = "${RELEASE_USER}" ]; then
    LOG_ROOT_DIR="/opt/data/log"     # release 用户走系统日志目录
else
    LOG_ROOT_DIR="${HOME}/data/log"  # 其他用户走自己家目录
fi

问题:

  • alvis 是老环境硬编码,新环境部署用户是 bigdata,迁移时必须一起改
  • "按执行者身份决定日志路径"把运行身份与路径策略耦合在一起,代码里到处都要判断当前用户
  • 调度执行(bigdata)和个人调试的日志散落到不同目录,排查问题时需要来回切换
  • 本质是把环境差异写进代码,而不是写进配置

建议:

  1. 删除 whoami == RELEASE_USER 分支逻辑
  2. 日志根路径统一由 conf/env.shLOG_ROOT_DIR 决定(默认 /opt/data/log),个人调试可在自己的 shell 里 export LOG_ROOT_DIR=~/data/log 覆盖
  3. RELEASE_USER 若仍需保留(如 publish.sh 发布身份校验),只作为白名单,不参与日志路径决策

7.3 部署改进

  • publish.sh 使用 re-all 命令(自定义的 SSH 分发脚本)全量同步
  • 建议:考虑引入版本化部署(tag + 软链接切换),便于回滚

7.4 DataX 限速逻辑

job_config_generator.py 中根据时间段(7:50-19:00 白天限速,其余时间放开)动态设置 DataX 传输速率:

if 750 < local_time < 1900:
    speed = self.get_speed(10, byte=10485760, record=40000)   # 白天低速
else:
    speed = self.get_speed()                                   # 夜间高速

建议:将时间段和速率配置化,避免硬编码。

八、重构优先级排序

阶段 任务 优先级
P0 模块重命名 tendata→dw_base、launch-pad→jobs
P0 清理所有业务代码(launch-pad 中保留的样本)
P1 硬编码提取到 conf/
P1 __init__.py 瘦身,拆分初始化逻辑
P1 敏感信息(Webhook token 等)移出代码
P2 __contains__in 全局替换
P2 删除废弃空模块和注释代码
P2 搭建 tests/ 基础框架 + UDF 单测
P2 精简 requirements.txt
P3 日志模块统一
P3 SQL 注入修复
P3 部署脚本改进