# 重构路线

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

## 〇、全景与依赖 DAG

线性 P0-P3 排序不适合实际推进 —— 配置外移、模块重组、基础设施可以并行，而业务 SQL / 老代码删除强依赖它们。本节把全部任务按**聚簇**组织，按**依赖边**而非阶段号推进。

### 聚簇划分

| 聚簇 | 名称 | 范围 | 对应旧阶段 |
|------|------|------|-----------|
| **A** | 配置外移 / 硬编码清理 | `conf/env.sh` / `workers.ini` / `alerter.ini` / `spark-defaults.conf` / `datax-speed.ini` / datasource 多环境 / DataX 路径解耦 | 阶段 2 主体 |
| **B** | `dw_base/` 重组 | B1 `__init__.py` 瘦身 · B2 `common/utils/io/ops` 四模块边界定稿 · B3 代码风格修正（`__contains__` / SQL 注入 / Shell-Python 重复） · B4 新占位模块 registry | 阶段 2 / 4 混合 |
| **C** | `bin/` 入口收口 | `datax-import` / `datax-export` 两命令收口 · `datax-gc-generator` 从零重写 · `csv-to-hdfs-starter` 实现 | 阶段 1 尾 + 阶段 2 |
| **D** | 基础设施 | `tests/` 测试体系 · 告警模块重写 · 日志模块统一 · `dq/` 数据质量 · `wiki/` Docmost · `pm/` 项目管理集成 | 阶段 2 / 4 |
| **F** | 老代码删除 | `launch-pad/` 整删 · 其他已确认废弃 | 阶段 5 |

> 新业务 SQL 从零开发（`jobs/{raw,ods,dwd,dws,tdm,ads}/` + DS 工作流）不属于重构 scope，不纳入聚簇。新业务 SQL 生产稳定一个完整周期后触发 F（`launch-pad/` 整删）。

### 依赖 DAG

```
┌──────────────┐
│ B2 四模块    │
│    边界定稿  │
└──────┬───────┘
       ↓
 ┌─────┴──────┐
 ↓            ↓
┌──────────┐ ┌──────────┐
│ B4 新占位 │ │  C bin   │
│  (骨架)   │ │  收口    │
└──────────┘ └──────────┘

A 全部子项 ─────┐
B1 __init__ 瘦身┤  ✅ 2026-04-21
A2 spark conf ─┤  ✅ 2026-04-21
B3 风格修正 ────┤ 可独立推进，与上图并行
D 基础设施 ─────┘

       (A + B + C 基本就绪 → 新业务 SQL 从零开发，不属于重构 scope)
                 ↓
              (新业务 SQL 生产稳定一个完整周期后)
                 ↓
              ┌────┐
              │ F  │  老代码删除
              └────┘
```

**关键依赖边**（强依赖，不能翻转）：

- **B2 → B4**：四模块（`common/utils/io/ops`）边界不定，新占位模块放哪里都是赌博
- **B2 → C**：`bin/` 两命令的底层实现要放 `dw_base/datax/entry.py`，属于四模块定稿后的延伸
- **新业务 SQL 生产稳定 → F**：`launch-pad/` 删除前置条件是新 `jobs/` 在生产跑稳一个完整周期（新业务 SQL 开发本身不纳入重构 scope）

**可并行的事**（无强依赖）：

- A 大部分子项（env.sh / workers.ini / alerter.ini / datax-speed.ini / datasource 多环境 / DataX 脚本路径解耦）彼此独立
- B3 代码风格修正（`__contains__` / SQL 注入）与任何聚簇都不冲突
- D 基础设施（tests 骨架 / dq / sync / pm）骨架已建，实现可滚动推进

### 本文各节与聚簇的对应

- §一 模块重命名 → **跨 B/F**（历史项，大部分已完成）
- §二 消除硬编码 → **聚簇 A + 部分 C**
- §三 `__init__.py` 瘦身 → **聚簇 B（B1）**
- §四 代码风格修正 → **聚簇 B（B3）**
- §五 清理废弃代码 → **聚簇 F**（目前已清零，留档案）
- §六 测试体系 → **聚簇 D**
- §七 其他建议 → **聚簇 D + A 杂项**

---

## 一、模块重命名（高优先级） [聚簇 B / F]

### 1.1 `tendata` → `dw_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.*", ...)` | ✅ 2026-04-22 已改为 `poyee-data-warehouse`（两处字面量：`dw_base/utils/file_utils.py:9` + `dw_base/utils/hdfs_merge_small_file.py:7`） |
| 目录名引用 | `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` §1 目录树与 `30-开发规范.md` §4.2。业务域代码统一使用命名规范定义的 `trd`/`usr`/`prd`/`shp`/`pub`/`dim`（见 `21-命名规范.md` §3.2）。

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

## 二、消除硬编码（高优先级） [聚簇 A + 部分 C]

### 2.1 当前硬编码清单

| 硬编码内容 | 所在位置 | 建议方案 |
|-----------|---------|---------|
| 项目部署目录 `poyee-data-warehouse/` | `bin/publish.sh`（2026-04-20 从根目录挪入 `bin/`） | 新项目发布目录为 `/home/bigdata/release/poyee-data-warehouse/` |
| `DATAX_WORKERS` + `DATAX_WORKERS_WEIGHTS` + `RELEASE_HOST` 硬编码 | `bin/common/init.sh:13-28` | ✅ 2026-04-23 整体移入 `conf/workers.ini`（`[release] host` + `[weights]` 两 section，ini 格式入库），init.sh 改为纯 bash 解析；新集群 hostname cdhmaster02 / cdhnode01-03 |
| `HADOOP_CONF_DIR='/etc/hadoop/conf'` / `SPARK_CONF_DIR='/etc/spark/conf'` / `PYSPARK_DRIVER_PYTHON='/usr/bin/python3'` / `PYSPARK_PYTHON='/usr/bin/python3'` | `__init__.py` | ✅ 2026-04-29 收口：HADOOP_CONF_DIR / SPARK_CONF_DIR 加 `${VAR:-default}` 落 `conf/env.sh`，由 `bootstrap_env` setdefault 注入；PYSPARK_DRIVER_PYTHON / PYSPARK_PYTHON 改用 `os.environ['PYTHON3_PATH']` 复用，消除双份硬编码 |
| 告警 Webhook（钉钉 / 企微 Key） | `dw_base/common/alerter_constants.py`（老告警模块已于 2026-04-20 删除，含 `dingtalk_notifier.py` / `ent_interface_dingtalk*` / `bin/dingtalk-work-alert.sh`） | 新告警模块重写时 Webhook Key 外移到 `conf/alerter.ini`（**入库**——部署靠 git pull，gitignore 会拉不到；webhook key 不算高敏感，最多被拿去发垃圾消息），Python 侧改 ConfigParser 加载；`alerter_constants.py` 整个删除；新项目不再使用钉钉 |
| Spark 默认参数（executor/driver/shuffle/sql.*） | `dw_base/spark/spark_sql.py` 构造函数 + `.config(...)` 链 | 移入 `conf/spark-defaults.conf`，SQL 文件可用 `SET` 覆盖，见 §2.3 |
| DataX ini 路径前缀剥离 `conf/datax/config/` | `bin/datax-single-job-starter.sh`（TEMP 处理）、`bin/datax-job-config-generator.py`（`replace('conf/datax/config/', '')`）、`bin/datax-multiple-job-starter.sh`（日志路径派生） | 原目录已整体挪到 `conf/bak/` 并 gitignore，脚本里 replace 现在是 no-op 死逻辑。去除前缀假设，改为靠 ini 文件名（= 任务唯一标识）识别用途 |
| DataX 生成 JSON 输出目录名 `conf/datax/generated` | `bin/datax-job-config-generator.py` 末尾 `default_output_dir`、`bin/datax-single-job-starter.sh` 第 89/118 行、`bin/datax-multiple-job-starter.sh` 第 187 行、`.gitignore` | 目录改名 `conf/datax-json/`；子路径扁平化为 `conf/datax-json/{ini_basename}.json`（去掉 src_dst / project_layer_env 等派生层级，依赖 ini 文件名全局唯一）；`.gitignore` 同步改 |
| JOB_NAME / JSON 文件名的 `ini→json` 转换逻辑重复实现 | Python 侧 `bin/datax-job-config-generator.py:126`（`os.path.basename(gcf).replace('.ini', '.json')`）+ Bash 侧 `bin/datax-single-job-starter.sh:88`（`basename .ini`） | 合一到 `dw_base.datax.path_utils.job_name_from_ini()`（或类似工具）；Bash 侧通过 `python3 -c` 调用或在 `bin/common/init.sh` 定义等价 shell 函数，单一来源 |
| ini 里 `dataSource` 字段拼接环境后缀 | 老项目写法 `dataSource = pg-hobby-prod` | 改为 `dataSource = {db_type}/{env}-{实例简称}`，代码按首段斜杠判 db_type（即父目录名）；source ini 落位 `datasource/{db_type}/{env}-{实例简称}.ini`，无 env 子目录。已于 2026-04-22 落地（`dw_base/datax/plugins/plugin.py:37`、`plugin_factory.py:34`） ✅ |
| 导出类 ini 扇出撞名风险 | `jobs/ads/{域}/` 下 ini 若都以源 Hive 表名命名，同一张 ads 表扇出到多个目标库时会重名覆盖 | 命名规则改为 `{源 Hive 表名}__{目标 db_type}_{目标 instance}.ini`（双下划线分隔源/目标） |
| `dw_base/common/template_constants.py` 大量死代码 | 定义了 20+ 个 SQL 模板路径常量，实际只有 2 个（`MYSQL_HIVE_CREATE_TABLE_TEMPLATE` / `MYSQL_HIVE_HBASE_CREATE_TABLE_TEMPLATE`）被引用，其余 18 个零 import | 整个文件删除；连带废弃下一条 ✅ 2026-04-29 整文件删 |
| `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/` ✅ 2026-04-29 两方法删 |
| 缺少集中的开发者参考模板目录 | —（新增） | 已建 `conf/templates/{datax/{datasource,sync},spark/{sql,ddl}}/`，模板用 `*.template.{ini,sql}` 双扩展名。与上条废弃的运行时模板完全不同：这里的模板不被任何代码读取，只供开发者对照写新文件；入口见 `kb/30-开发规范.md §6` |

### 2.2 建议的配置结构

```
conf/
├── env.sh                    # Shell + Python 环境变量单源（Python 侧由 dw_base/utils/env_loader.py 通过 bash 子进程解析注入 os.environ）
├── workers.ini               # DataX Worker 列表与权重
├── alerter.ini               # 告警 Webhook 配置（入库；见 §2.1）
├── spark-defaults.conf       # Spark 底层行为/开关类（11 条，初始化后少改；Spark 原生格式）
└── spark-tuning.conf         # Spark 资源/并行度/队列类（10 条，业务早期常改；同 tuning 相同 key 覆盖 defaults）
```

### 2.3 Spark 配置三级覆盖策略（已完成）

**2026-04-21 落地形态**：按业务调整频率拆两文件入库 + `spark_sql.py` 三级覆盖。

**两文件拆分**：

- `conf/spark-defaults.conf`（11 条）—— 底层行为/开关类，初始化后少改（`spark.sql.adaptive/broadcastTimeout/codegen/arrow*/files/statistics.*` + `spark.dynamicAllocation.enabled` + `spark.files.ignoreCorruptFiles` + `spark.debug.maxToStringFields` + `spark.port.maxRetries`）
- `conf/spark-tuning.conf`（10 条）—— 资源/并行度/队列类，业务早期常改（`spark.{driver,executor}.{memory,cores}` + `spark.executor.instances` + `spark.executor.memoryOverhead` + `spark.driver.maxResultSize` + `spark.default.parallelism` + `spark.sql.shuffle.partitions` + `spark.yarn.queue`）

两文件都用 Spark 原生 `key value` 格式（空白分隔、`#` 注释、无 section），与 `spark-submit --properties-file` 同语法。

**三级覆盖**：

```
L1   conf/spark-defaults.conf  +  conf/spark-tuning.conf     （相同 key tuning 覆盖 defaults）
     ↓
L2   SQL 文件内 SET spark.xxx=yyy     （query() 预扫描塞入 _final_spark_config，session 启动前写入 builder）
     ↓
L3   SparkSQL(...) 显式传参  +  extra_spark_config  +  命令行 -sc
```

**与仓库改名的联动** ✅ 2026-04-22：仓库改名 `tendata-warehouse-release` → `poyee-data-warehouse` 已完成（项目根由用户手动改名 + 路径正则两处字面量同步）。`.idea/*.iml` + `modules.xml` 因 `.idea` + `*.iml` 在 `.gitignore`，老 iml / modules.xml 引用 / workspace.xml module name 残留是本地 IDE 状态，不入库亦不影响运行，不处理。

### 2.5 DataX 脚本路径解耦

**现状**：`bin/datax-single-job-starter.sh` 和 `bin/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` 的位置——能跑但与新约定不符。

**目标态**：清理死路径派生逻辑，JSON 输出扁平化为 `conf/datax-json/{ini_basename}.json`（依赖 ini 文件名全局唯一），日志目录简化为 `${LOG_ROOT_DIR}/datax/${START_DATE}/${JOB_NAME}.log`。

**已作废的子项**（2026-04-22 方向反转）：`-env` 参数 + datasource 按 env 分子目录 + 一套代码跑多环境。前期跨环境同步是常态（test 业务库 → prod HDFS），不存在"全局 env"概念，sync ini 里 `dataSource = {db_type}/{env}-{实例简称}` 显式指向具体 source ini；source ini 落位 `datasource/{db_type}/{env}-{实例简称}.ini`，扁平组织。代码侧 db_type 判定改按首段斜杠（= 父目录）在 `plugin.py:37` + `plugin_factory.py:34` 落地（✅ 2026-04-22）。

### 2.6 DataX 入口脚本收口为两条命令 ✅ 2026-04-23

> ✅ 2026-04-23 落地：新入口 `bin/datax-{hive-import,hdfs-export}-starter.{sh,py}` + `dw_base/datax/` 7 模块（`path_utils` / `worker` / `partition` / `runner` / `batch` / `entry` / `cli`）；老 7 个脚本（`datax-{single,multiple,multiple-hive}-job-starter.{sh,py}` + `datax-job-config-generator.py`）整体删除。详见 kb/92 对应日 changelog。本节保留为上下文（kb/90 retention 规则）。

**现状**：DataX 入口分散，学习成本高：

- `bin/datax-single-job-starter.sh` — 单 ini 执行
- `bin/datax-multiple-job-starter.sh` — 批量 + 按 start-date/stop-date 范围展开
- `bin/datax-multiple-hive-job-starter.sh` — Hive 表批量（语义与上者重合）
- `bin/datax-single-hdfs-job-starter.sh`（如有残留） — 单次 HDFS 导出
- `bin/datax-job-config-generator.py` — ini → json 翻译器（内部工具）
- `bin/datax-gc-generator.py` — ini 元生成器（详见 §2.7）

**目标态**：顶层收成两个命令，命名锁定 `bin/datax-hive-import-starter.sh` / `bin/datax-hdfs-export-starter.sh`（对齐现有 `spark-sql-starter` / `datax-*-starter` 命名风格）；参数按老入口一比一平迁、不引入新能力；底层 worker 选择 / 日志路径 / json 翻译下沉到 `dw_base/datax/` 包内，调用方不感知。

| 顶层命令 | 语义 | 参数集（老入口平迁） |
|---|---|---|
| **`bin/datax-hive-import-starter.sh`** | 目标=Hive（reader=外部 / writer=hdfs + 自动分区），对应 `jobs/raw/` | `-ini <file>` 单 ini · `-inis <dir>` 目录批量 · `-start-date / -stop-date` · `-host / -random` · `-parallel` · `-skip-datax` · `-skip-partition`（默认开启分区管理）· `-t <db.table>` 显式追加需建分区的表 |
| **`bin/datax-hdfs-export-starter.sh`** | 源=HDFS（reader=hdfs / writer=外部如 ES/Mongo/MySQL/Kafka），对应 `jobs/ads/` | 同上减 `-skip-partition` / `-t`；源路径存在性 check 沿用老 `check_data_exists` 默认 fail-fast 行为，不暴露开关 |

**老参数不平迁**（新入口移除）：

- `-c / -cd / -jc / -jcd`：json 输入形态不再对外暴露，入口自动调 ini→json 翻译
- `--override` + 5 个脚本内 `xxx_array=()`（`partitioned_tables` 等）：数组在老 `datax-multiple-hive-job-starter.sh` 中本就是未激活的空壳（2026-04-23 查证），一并删除

**实现建议**：

1. 新增 `dw_base/datax/{entry,runner,batch,worker,partition,path_utils}.py` 6 个模块，把老 shell 脚本里的 worker 选择 / 批量展开 / 日志路径派生 / ini→json 翻译 / Hive 分区管理逻辑全部搬迁，两个 sh 只做 bash 环境初始化 + 参数解析 + 调 python
2. 模块放 `dw_base/datax/` 包内，不横跨 `io/` / `utils/` —— **放宽聚簇 B2 前置**（四模块边界定稿前先落地）。等 B2 定稿后，`path_utils.log_path` 等纯函数、`worker.ssh_run` 中跨包部分再挪到 `dw_base/utils/` / `dw_base/io/`，只改 import 路径
3. `-inis <dir>` 目录扫描只递归 `.ini` 文件
4. 老脚本 `datax-{single,multiple,multiple-hive}-job-starter.{sh,py}` + `datax-job-config-generator.py` 在冒烟 2（新入口端到端通过）后整体删，**不保留兼容转发封装**
5. `partition.py` 预建分区的 dt 用 `start_date`（业务日），和 HDFS writer（见 `hdfs_writer.py:23-26`）对齐；分区内允许含次日漂移数据（按 ADR-03 raw 不纠正分区漂移，配套 raw 48h 宽窗机制）。早先方案曾用 `stop_date - 1 day`——单日窗口下与 start_date 数值相等没问题，但与 raw 宽窗（stop = 业务日+2）配合时分区会跑到业务日+1 与业务日错位，反向校正回 start_date

**本轮不做、后延 ADR**：下列能力属新增需求、非老入口平迁，暂不实现；若将来出现明确场景，单独开 ADR 落 `kb/93`。按优先级分：

**高优先级**（正式 raw 库上线前需要）：

- ✅ 2026-04-23 **DataX 字段级脱敏（合规硬约束：敏感值必须不出业务库）**：规范做法 = ini 里 `[mask]` 声明式段（`cert_birthday = month_trunc` 等），代码侧 `dw_base/datax/mask.py` 翻译为 PG `querySql` 内 SQL 表达式下发源库执行，敏感原值从未离开业务库。PG 方言当前支持 5 种脱敏（静态：`month_trunc` / `md5` / `mask_middle`；动态：`keep_first_{n}` / `keep_last_{n}`）。优先级 `手写 querySql > [mask] > table`，手写用于复杂场景 override。**不走** DataX 层 transformer / reader `transform =` 方案——这些路径是原值先进 DataX JVM 再变换，等同"已出业务库再脱"，合规退步，本项目不采纳。MySQL / ClickHouse 等其他 reader 按需扩展 `mask.py` 顶层字典

**低优先级**（有明确批量回刷场景再做）：

- 日期范围自动按日展开 + N 个 json 分发多 worker（`-start-date 20260401 -stop-date 20260410` → 自动切 10 份 json，每份独立选 worker）——**可能不做**，按天补数职责归 DolphinScheduler 更合理（见 `kb/93 ADR-01` 草案）
- **DataX 仅本机执行**（砍 ssh 远端分发 + `workers.ini` + `select_worker`）：分布式分发由 DolphinScheduler worker group 承担，DataX 不重复随机（见 `kb/93 ADR-02` 草案）
- **`plugin_factory` 解耦**：当前工厂模块顶层 eager import 所有 reader/writer（mysql/pg/mongo/hdfs/es/kafka/hbase/ch/...），任一 plugin 的 top-level import 破损就连累整个 DataX 链路（2026-04-23 已踩：`dw_base/database/` 删除后 `mysql_reader` 的 `MySQLColumn` import 挂，挂一扯一串）。改造方向二选一：(a) lazy import：`importlib.import_module(f"dw_base.datax.plugins.reader.{db_type}_reader")` 按需加载；(b) registry pattern：每个 reader 用 `@register("postgresql")` 装饰器自注册，工厂解耦于具体 plugin 列表
- ✅ 2026-04-23 **ini 配置化脱敏** 已落地 `dw_base/datax/mask.py`（PG 5 种脱敏），详见高优先级"DataX 字段级脱敏"条

**第三条命令 `datax-gc-generator`（ini 元生成器）独立保留**：用户已确认。职责是"从 PG 扫 schema 生成 ini 参考模板"，和"执行 ini"不是一回事，不收口到上面两条里。详见 §2.7。

### 2.7 `datax-gc-generator` 从零重写 ✅ 2026-04-29

**现状**（凭查证 2026-04-18）：`bin/datax-gc-generator.py` 支持 `from ∈ {mysql, hdfs}` × `to ∈ {elasticsearch, hbase, hdfs, kafka, mongo, mysql}`，覆盖面大、代码沉重，且：

- **全项目没有任何其他模块 import 或 shell 调用它**（仅 3 处 `SparkSQL('datax-gc-generator')` 是自己设置 app name）
- 内部走 `MySQLHandler` + `MySQLDataSource` + `convert_mysql_column_types` + `MySQLReader.generate_hive_ddl()` 一条链，与 §2.1 已登记废弃的"自动 DDL 生成"方向相冲突
- 新项目新业务只需要 `from=pg to=hdfs` 一条路径，其他组合全是老项目遗产

**定位**：**参考模板生成器**，不是"一键出可用 ini"。产物是开发者人工调整的起点 —— 常见修改包括字段剪裁（只同步用到的列）、WHERE 过滤条件、hivePartitions 配置、大表拆分策略等。开发者 diff 参考模板和自己的需求，改完再把成品 ini 提交到 `jobs/raw/{domain}/`。

**方向**：整个文件废弃 + 从零重写 ✅ 2026-04-29 重写为 `bin/datax-sync-template-gen.py`（PG → HDFS sync 模板生成器，全字段参考模板，pg8000 直连查 schema，-o 三态：stdout / workspace 默认 / 自定义目录）

**重写目标**：
- 仅支持 `from=pg to=hdfs`
- 读 PG 表结构（`information_schema.columns` + `pg_catalog.pg_description` 取字段注释）
- 输出到**开发者本地的 `workspace/{yyyymmdd}/{name}.ini`**（`workspace/` 被 `.gitignore` 排除，**不入仓**），作为参考模板（原始全字段 + 默认分区 + 默认 `where=1=1`）；开发者裁剪后再把最终 ini 提交到 `jobs/raw/{domain}/`
- **不再生成 DDL**：DDL 统一由开发者按 `21-命名规范.md` 手写到 `manual/ddl/{layer}/{domain}/`（CLAUDE.md 单一来源约束，与 §2.1 已登记项一致）

**目录约定**：
- `workspace/` 在仓库根，**仅存在于开发者本地**，整个目录进 `.gitignore`
- `workspace/{yyyymmdd}/` 按运行日期分子目录，便于开发者看"我今天生成了哪些候选"
- 与 `manual/imports/{yyyymmdd}/` 的分工：`manual/imports/` 放一次性**执行**的 SQL / ini（会入仓做审计证据，执行完归档），`workspace/` 放自动化工具**未经人工确认**的中间产物（永不入仓）

**拆除清单**（实际执行时 scope 收紧——mysql 整体作为运行时同步代码保留，仅删纯 generator 死代码）：
- `dw_base/database/mysql_utils.py` 的 `list_tables` / `list_columns` 方法 ✅ 整 `database/` 目录已于 2026-04-20 老业务批清理删除
- ~~`dw_base/datax/datasources/mysql_data_source.py`~~ **保留**（mysql 同步运行时 DataSource，与 PostgreSQLDataSource 同等地位；未来 mysql 同步可能用）
- `dw_base/datax/plugins/reader/mysql_reader.py` 的 `generate_hive_ddl` / `generate_hive_over_hbase_ddl` 方法 ✅ 2026-04-29 删
- `dw_base/datax/datax_utils.py` 的 `convert_mysql_column_types` ✅ 2026-04-29 整文件删（零外部引用）
- 所有 mongo / kafka / hbase writer 在 generator 里的分支 ✅ 2026-04-29 随 generator 整文件删

**注意**：以上删除范围与 `dw_base/datax/plugins/` 里仍在被真实采集任务调用的 reader/writer 不冲突 —— 真实采集任务只用到 `HDFSReader` / `HDFSWriter` / `MongoWriter`（如果还有 mongo 采集任务）。删之前要用 `grep -r "from dw_base.datax.plugins.reader.mysql_reader"` 再确认一次。

### 2.8 HDFS 数据源 ini 支持 HA nameservice（**Path B 锁定**）

**结论（凭 2026-04-18 新 CDH 环境实测）**：DataX hdfswriter 在新环境下连 HA 集群**必须**在 json 里显式提供 `hadoopConfig` 块，这是 §2.8 改造的动机。

**为什么 `HADOOP_CONF_DIR` 对 DataX 无效（实测三连）**：

| 测试 | json 内容 | `HADOOP_CONF_DIR` | 结果 |
|---|---|---|---|
| 1 | `defaultFS` + 完整 `hadoopConfig` | 未设置 | ✅ 成功 |
| 2 | 只有 `defaultFS` | 未设置 | ❌ UnknownHostException: nameservice1 |
| 3 | 只有 `defaultFS` | `/etc/hadoop/conf` | ❌ 仍 UnknownHostException |

`hadoop fs -ls` 命令能解析 `nameservice1`，是因为 `hadoop` shell 脚本把 `$HADOOP_CONF_DIR` **加入了 Java classpath**（`hdfs-site.xml` 才能被 `Configuration` 自动加载）。`datax.py` 启动 Java 时**不做这件事**，classpath 里不含 `/etc/hadoop/conf`，所以 DataX 的 Hadoop `Configuration` 对象是"干净的"，只能靠 json `hadoopConfig` 块显式注入 HA 参数。

（老项目在上一个公司/服务器只写 `defaultFS` 也能跑通 HA，最可能的原因是运维把 `hdfs-site.xml` 塞进了 DataX 的 classpath 目录 —— 比如 `datax/conf/` / `datax/lib/` / 某 plugin 的 `libs/`。新环境 `/opt/datax` 下没有这类预置文件，不走这条路。）

**`dw_base/__init__.py` 的 `HADOOP_CONF_DIR` + `HIVE_CONF_DIR` export 仅对 Spark 入口生效**：(a) `HADOOP_CONF_DIR`：Spark on YARN 启动时 `SparkSubmitArguments.validateSubmitArguments` 会强校验该 env（不设直接抛 `When running with master 'yarn' either HADOOP_CONF_DIR or YARN_CONF_DIR must be set`，2026-04-20 实测）。(b) `HIVE_CONF_DIR`：Spark `.enableHiveSupport()` 找不到 `hive-site.xml` 时会静默回落 in-memory metastore——session 能起但看不到 HMS 真实库表（症状：`describe test.xxx` 在 Hive CLI 有，在 Spark 里 `Table or view not found`，2026-04-20 实测）。两者对 DataX JVM 子进程都无影响——一是 classpath 不含 conf 目录（见上），二是 DataX 由 `datax-single-job-starter.sh` 通过 `python3 datax.py` 启动，并未 import `dw_base`；DataX HA 靠 ini `[hadoop_config]` 节显式注入。早先 2026-04-18 曾把这两行当"死代码"删除（`kb/92:158`），仅从 DataX 视角判断忽略了 Spark 路径，2026-04-20 分两批恢复（HADOOP_CONF_DIR 在先，HIVE_CONF_DIR 补齐）。

---

**正确的 json 形态**：

```json
"writer": {
  "name": "hdfswriter",
  "parameter": {
    "defaultFS": "hdfs://nameservice1",
    "hadoopConfig": {
      "dfs.nameservices": "nameservice1",
      "dfs.ha.namenodes.nameservice1": "nn1,nn2",
      "dfs.namenode.rpc-address.nameservice1.nn1": "192.168.33.61:8020",
      "dfs.namenode.rpc-address.nameservice1.nn2": "192.168.33.62:8020",
      "dfs.client.failover.proxy.provider.nameservice1": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
    },
    ...
  }
}
```

**ini 新格式**：

**非 HA / 单 NN（dev / test 环境）**：

```ini
[base]
defaultFS = hdfs://192.168.33.61:8020
```

**HA / nameservice（prod 环境）**：

```ini
[base]
defaultFS = hdfs://nameservice1

[hadoop_config]
dfs.nameservices = nameservice1
dfs.ha.namenodes.nameservice1 = nn1,nn2
dfs.namenode.rpc-address.nameservice1.nn1 = 192.168.33.61:8020
dfs.namenode.rpc-address.nameservice1.nn2 = 192.168.33.62:8020
dfs.client.failover.proxy.provider.nameservice1 = org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider
```

**设计要点**：
- `[hadoop_config]` 节的 key **原样照搬 `/etc/hadoop/conf/hdfs-site.xml`**，不做缩写 / 翻译 —— 运维直接复制粘贴，出了问题可以字段级对比，零心智成本
- `[hadoop_config]` 节**可选**：不写即视为非 HA，代码不注入 `hadoopConfig`，生成的 json 只有 `defaultFS`，与单 NN 环境兼容
- 将来若要支持多 nameservice（federation），`[hadoop_config]` 节天然兼容（多写几组 key 即可），无需再改 ini schema

**代码改造清单**：

| 文件 | 改动 |
|------|------|
| `dw_base/datax/datasources/hdfs_data_source.py` | 覆写 `get_datasource_dict()`：父类逻辑外，检测 `[hadoop_config]` 节是否存在；存在则把整节作为 dict 塞进 `ds_dict['hadoopConfig']` |
| `dw_base/datax/plugins/plugin.py` | **不用改** —— `load_data_source()` 的 `for key, value in ds_dict.items()` 里，value 是 dict 时走 Python 原生赋值，json 序列化自然成立 |
| `dw_base/datax/plugins/reader/hdfs_reader.py` / `writer/hdfs_writer.py` | **不用改**，`defaultFS + hadoopConfig` 由 `load_data_source()` 自动注入到 `parameter` |
| `dw_base/__init__.py` | **保留** `os.environ['HADOOP_CONF_DIR']`（Spark on YARN 启动强校验）+ `os.environ.setdefault('SPARK_CONF_DIR', '/etc/spark/conf')`（pip pyspark 默认指向自身空 conf/，需显式指到集群配置才能加载 hive-site.xml）；DataX JVM 不读这些，HA 靠 ini `[hadoop_config]`。详见 `01-运行环境.md §4` |
| `bin/datax-gc-generator.py` | §2.7 从零重写时一并处理，这里不单独列 |
| `datasource/hdfs/{env}-{实例简称}.ini` | 按新 schema 新建（HA 集群带 `[hadoop_config]`；单 NN 只写 `[base] defaultFS`） |

**回归测试**：
- prod HA 集群：新 ini 跑一次真实 PG→HDFS 任务，生成 json 含完整 `hadoopConfig`，任务成功（2026-04-18 已用手写 json 预验证）
- NameNode 主备切换期间跑一次，DataX 自动切到 standby（HA 的原始动机）
- dev / test 单 NN 集群：不写 `[hadoop_config]`，生成 json 只含 `defaultFS`，任务成功

### 2.9 DataX 速率控制外移到 conf（中优先级）

**现状**：`dw_base/datax/job_config_generator.py:60-67` 硬编码分时调度：

```python
local_time = int(datetime_utils.formatted_now('%H%M'))
if 750 < local_time < 1900:
    speed = self.get_speed(10, byte=10485760, record=40000)          # 白天 10 channel × 10 MB/s
    core_speed = self.get_core_speed(byte=10485760, record=40000)
else:
    speed = self.get_speed()                                          # 夜间 6 channel × 256 MB/s
    core_speed = self.get_core_speed()
```

**问题**：
- 白天 / 夜间的边界（0750、1900）、channel 数、单 channel 速率全写死在代码里，想调一个值要改 py 源码重新发布
- 不同业务域 / 不同表的速率诉求可能不同，现在一刀切
- 生产偶发突发（白天要抢刷一次全量）没有临时放大速率的口子
- 硬编码的动机在代码里一字没留（为什么 0750-1900？避开业务高峰？避开实时链路？后人读代码猜不到）

**目标**：把速率配置抽到 `conf/datax-speed.ini`，`JobConfigGenerator` 运行时读；默认值 = 现硬编码值，保持向后兼容。

**conf 格式**（ini，按时间段分段，自上而下匹配，未命中走 `[default]`）：

```ini
; conf/datax-speed.ini
; 速率档位定义：从上到下依次匹配，第一条命中的为准；无匹配走 [default]

[daytime]
hours = 07:50-19:00
channel = 10
byte_per_channel = 10485760       ; 10 MB/s/channel，避开业务高峰
record_per_channel = 40000

[default]
channel = 6
byte_per_channel = 268435456      ; 256 MB/s/channel，夜间放满
record_per_channel = 100000
```

**代码改造**：
- 在 `JobConfigGenerator.__init__()` 读 `conf/datax-speed.ini`
- 用当前时间在各段 `hours` 里找匹配，取 `channel / byte_per_channel / record_per_channel`
- `assemble()` 里的分支逻辑改为 `speed = self.resolve_speed_profile()` 一行
- 落 `conf/datax-speed.ini` 时在文件头注释里把**"白天是为了避开业务高峰"**的动机写清楚（消除默认值的来历盲区）

**扩展方向**（P2，先不做）：
- 支持按 ini 名（业务表）在 conf 里覆盖特定任务的速率
- 支持命令行 `-speed fast` / `-speed slow` 手动切档（突发高峰 / 限流时用）

### 2.10 `dw_base/` 四模块边界（common / utils / io / ops） [聚簇 B（B2）]

**背景**：老项目 `dw_base/` 下子目录职责混乱 —— `utils/` 里塞过 Excel/Hive 读写、小文件合并；`database/` 又和 `datax/datasources/` 重叠；缺少 I/O 边界和"数据湖运维"这类长期职责的收纳点。重组把全部非业务代码（非 `spark/` / `udf/` / `datax/` 等明确功能域）按**边界属性**归入四个顶层模块，规则如下：

| 模块 | 边界属性 | 典型内容 | 反例（不该放在此处） |
|------|---------|---------|---------------------|
| **`common/`** | 常量 / 全局上下文 / 无行为的元数据 | 颜色常量、枚举、模板路径常量、项目级单例 context | 任何"动词"型函数 —— 那是 `utils/` |
| **`utils/`** | 纯函数，无外部副作用，可纯 Python 单测 | 日期格式、字符串切分、配置解析、数据结构转换 | 读/写 DB / 文件 / HDFS —— 那是 `io/` |
| **`io/`** | 与外部系统通信的**边界** | DB connector 薄封装、CSV/JSON/Excel 读写、HDFS 文件读写 | 数据湖维护动作（合并小文件、清理分区）—— 那是 `ops/` |
| **`ops/`** | 数据湖维护动作，会改变湖里的物理状态 | 小文件合并、分区清理、统计信息刷新、存储压缩 | 一次性的业务查询脚本 —— 那走 `jobs/` 或 `manual/` |

**落地规则**：

1. **写新代码前先对号入座**：如果一个函数既读文件又做纯计算，按**最强副作用**归类（读文件 → `io/`，即便函数里 90% 是计算）
2. **`io/` 只做薄封装**：不把业务 schema 嵌入 `io/`（老 `mongodb_utils.py` 把公司名查询嵌进去就是反例），业务 schema 留在调用方
3. **`ops/` 里的函数预期被多处复用**：只被一张表调用一次的"清分区"走 `manual/adhoc/`，不进 `ops/`
4. **跨模块依赖方向**：`ops/` 可依赖 `io/` + `utils/` + `common/`；`io/` 可依赖 `utils/` + `common/`；`utils/` 只依赖 `common/`；`common/` 不依赖任何项目内模块。**禁止反向依赖**

**B2 先于 B4 / C**：新占位模块（io/ops/pm/dq/sync）和 `bin/` 收口都假设四模块边界已定；边界不定就先别新建骨架，否则返工成本高。

### 2.11 新占位模块 registry（B4） [聚簇 B（B4）]

2026-04-20 本批建立的占位模块（空骨架，实现待后续阶段），登记在此便于跨会话追踪：

| 模块 | 路径 | 状态 | 何时实现 | 对应阶段 |
|------|------|------|----------|---------|
| `io/` | `dw_base/io/{db,file,hdfs}/` | 骨架 | B2 四模块边界定稿后，搬入 db/file/hdfs 读写封装 | 阶段 2 |
| `ops/` | `dw_base/ops/` | 骨架 | 阶段 4 随"重新实现小文件合并 / 分区保留工具"落地 | 阶段 4 |
| `pm/` | `dw_base/pm/` | 骨架 | 待确认 TAPD vs Jira 后实现（commit → 任务 ID 联动） | 阶段 4 之后 |
| `dq/` | `dw_base/dq/` | 骨架 | 待确认告警落点 + 首批规则后实现 | 阶段 4 |
| `wiki/` | `dw_base/wiki/` | 骨架 | 待确认 Docmost API 鉴权 / webhook 支持后实现 | 阶段 4 之后 |
| `tests/` | `tests/{unit,integration}/` | 骨架 | 阶段 4 首批 UDF 单测启动时同步补 `conftest.py` + fixtures | 阶段 4 |

### 2.12 通用 UDF 自查表（B 延伸） [聚簇 B]

**现状**：`dw_base/udf/common/spark_common_udf.py` 462 行，其中 `@udf` 装饰的注册函数 13 个（启动日志 `注册 Python UDF xxx` 条数即真值），另有若干普通 `def` 辅助函数不进 SparkSQL 注册表。函数注释已由开发者手动补完（编号 `UDF-XX` 顺序注入、分类分节 JSON / ARRAY / STRING / NUMERIC-DATE-HASH / CROSS-TYPE），本节不再规划"注释补齐"动作。

**问题**：未来新增通用 UDF 如果没有登记表，规模大了之后"这个函数有没有 / 叫什么 / 谁加的"全靠 grep。

**目标态**：

- **UDF 自查表**：`kb/31-UDF手册.md`（与 `30-开发规范.md` 同级独立文档）。表头 `函数编号 | 函数名 | 分类 | 入参 | 返回 | 描述 | 示例`，函数编号沿用 `spark_common_udf.py` 中的 `UDF-XX` 顺序注释。common 与 business 两类 UDF 都在本表登记，分两段（§1 通用 / §2 业务）。初版 13 个通用 UDF 已登记（2026-04-21）。

**回归检验**：
- 任意 SQL 文件直接 `SELECT my_udf(col)` 能跑通（common auto-load 链路未变）
- 自查表 §1 行数 = `spark_common_udf.py` 中 `@udf` 注册函数数（启动日志中 `注册 Python UDF` 的条数即真值）

**与其他条目的关系**：
- 2026-04-20 UDF 模块重组（kb/92）已完成重组动作，本节是其延伸（登记表）
- 不动 auto-load 机制（`bin/spark-sql-starter.py` + `dw_base/__init__.py:29 COMMON_SPARK_UDF_FILE` 常量），只补文档

## 三、`__init__.py` 瘦身（已完成 · 修剪式） [聚簇 B（B1）]

**2026-04-21 完成形态**：原地修剪，不拆 `core/`。`dw_base/__init__.py` 从 127 行 → 83 行。

**落地动作**：
- 删 `import findspark` + `findspark.init()`：pip pyspark 已在 venv site-packages 直接覆盖 `import pyspark`，findspark 在 CDH 节点上即便间接反推出 `$SPARK_HOME` 也只是把同版本 parcel pyspark 前插进 sys.path，业务表现零差异（冒烟已确认）
- 删 21 个外部零引用的颜色常量：`CHG_BOLD` / `NORM_BLU` / `NORM_WHT` / 7×`BOLD_*` / 7×`BGRD_*`（if/else 两分支同步删）
- 删 `IS_RUN_BY_NORMAL_USER` 状态变量（两处赋值外部均无引用）

**未拆分为 `core/env.py` / `core/colors.py` / `core/spark_env.py` 的理由**：findspark 删除后"findspark 按需 import"的懒加载诉求大半消失；剩余颜色/env 检测拆三文件需改 11 处调用点 import，收益比修剪方案差，暂不做。若后续单元测试对 "import dw_base 零副作用" 有刚性需求再拆。

**最终 `__init__.py` 保留**：`PROJECT_ROOT_PATH` / `PROJECT_NAME` / `bootstrap_env()` / `sys.path.append` / `HADOOP_CONF_DIR` / `SPARK_CONF_DIR` / `PYSPARK_*` / `COMMON_SPARK_UDF_FILE` / `BANNED_USER` 退出守卫 / `IS_RUN_BY_RELEASE_USER` / `IS_RUN_IN_RELEASE_DIR` / 启动 print / 6 个实际被引用的颜色常量（`DO_RESET` / `NORM_RED` / `NORM_GRN` / `NORM_YEL` / `NORM_MGT` / `NORM_CYN`）。

**2026-04-24 追加清理**：`cow_says()` 函数（os.system source bin/common/functions.sh）已删——bin 层 bash 壳整体清理时确认其为 no-op（子 shell source 后立即退出，无任何可见效果）。

## 四、代码风格修正（中优先级） [聚簇 B（B3）]

### 4.1 `__contains__` 反模式

全项目大量使用：
```python
if config.__contains__(key):       # 反模式
if self.REGISTERED_UDF.__contains__(name):
```

应改为：
```python
if key in config:                  # Pythonic
if name in self.REGISTERED_UDF:
```

### 4.2 Shell / Python 重复逻辑

`bin/common/init.sh` 和 `dw_base/__init__.py` 有大量重复的环境检测逻辑（用户判断、路径判断、日志目录、颜色常量）。

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

## 五、清理废弃代码（中优先级） [聚簇 F]

截至 2026-04-20 本节已无待清理项。后续若发现新的废弃代码，在下方表格追加登记；已完成项保留在"历史档案"表中留档。

### 5.1 待清理（当前为空）

| 模块/文件 | 状态 | 建议 |
|----------|------|------|
| _（无）_ |  |  |

### 5.2 历史档案（已完成）

| 模块/文件 | 完成动作 | 完成时间 / 来源 |
|----------|---------|----------------|
| `dw_base/{validation,ml,flink,elasticsearch}/` 四空壳模块 | 整目录删（仅 56 字节空 `__init__.py`，零引用） | 2026-04-20，见 `92-进度.md` |
| `dw_base/database/mongodb_utils.py` | 整文件删（184 行 = 19 行 MongoClient 薄包装 + 165 行注释，零外部引用，需要时 `MongoClient(uri)` 一行重写） | 2026-04-20，见 `92-进度.md` |
| `dw_base/database/` 整目录 | 整目录删（`__init__.py` + 仅剩的 `mysql_utils.py`，zero 外部引用；mysql 读写后续在 `dw_base/io/db/` 下重建） | 2026-04-21 |
| 5 个占位模块 README（`dq/` `io/` `ops/` `pm/` `sync/`） | 批量删，占位模块 docstring 即文档，不再要求单独 README | 2026-04-21 |
| `kb/README.md` | 整文件删，内容合入项目根 `README.md`（项目现状速读 / 文档索引 / 阅读建议三节），CLAUDE.md 引用同步更新 | 2026-04-21 |
| `conf/datax/` 下老项目遗留 ini / datasource 样例 | 整批挪入 `conf/bak/datax/{config,datasource}/`，由 `.gitignore:6 conf/bak` 拦截不入库 | 项目初始化 `8d2ade5`（2026-04-17） |
| `bin/flume-control.sh` | 整文件删（194 行）。职责：管理单个 Flume-Kafka-HDFS 作业生命周期（log/monitor/start/status/stop/restart/start-all/stop-all/restart-all 一组子命令）。实际状态：顶部 shebang 损坏（`ho#!/bin/bash`）、依赖已删除的 `bin/wechat-work-alert.sh`、L64 与 L162 `conf/flume/*.properties` vs `conf/flume/config/*.properties` 路径自相矛盾，整体已不可跑。按需重建时参考：`conf/flume/config/*.properties` 为作业配置约定、`flume-ng agent --conf-file` 为启动命令、启动前需 `mkdir -p ${LOG_ROOT_DIR}/flume-agent/${TODAY}`、监控循环 `head -n 1000 log | grep -E "gz failed\|java.io.IOException\|org.apache.flume.ChannelException\|java.lang.IllegalStateException"` 命中则发企微告警。Kafka→HDFS 接入通道重建时按新 `conf/alerter.ini` 告警外配重写，不沿用老 SKB_LITTLE_CUTE / 手机号硬编码 | 2026-04-21，见 `92-进度.md` |

> 代码里残留的 `conf/datax/config/` replace 死逻辑 + `conf/datax/generated` 默认值，属于 §2.x 路径硬编码清理范畴（改名为 `conf/datax-json/` + 删 replace），不在本节。

## 六、测试体系搭建（中优先级） [聚簇 D]

### 6.1 现状

- 仅 UDF 有少量 pytest 测试
- 核心模块（SparkSQL、DataX 配置生成）无测试
- 无 CI/CD 集成
- 2026-04-20 已建 `tests/{unit,integration}/` 骨架 + README（`.gitkeep` 占位，无实测用例），见 §2.11 占位模块 registry

### 6.2 建议的测试结构

```
tests/
├── conftest.py                    # pytest 公共 fixtures
├── unit/
│   ├── test_udf_common.py         # 通用 UDF 单测（纯函数，不依赖 Spark）
│   ├── test_udf_business.py       # 业务 UDF 单测（如有，按文件组织）
│   ├── 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 工作流中加入校验节点

## 七、其他建议 [聚簇 D + A 杂项]

### 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 / 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 / pypinyin —— 都只被即将清理的老业务代码引用 |

**后续事项**：

- LAZY 类依赖关联的老代码：`get_oldmongo_*` / `mg2es/` / `ent_interface_dingtalk*` 于 2026-04-20 第一批提前清理；同日第二批清理 `dw_base/oss/` 整目录、`dw_base/scheduler/` 整目录（含 polling_scheduler / drop_partitions / drop_daily_full_snapshot_tbls）、`dw_base/hive/` 整目录、`dw_base/utils/` 7 文件（data_distinct / diff_utils / excel_to_hive_utils / hive_diff_database / hive_to_excel_utils / pdt_check_table\*）；剩余 `customs/similarity.py` 等在阶段 4 / 阶段 5 一并清理
- 不需要 `requirements-base.txt` / `requirements-dev.txt` 分文件——当前依赖规模下单文件已经足够
- pyspark==2.4.0 固定（对齐 CDH 6.3.2 parcel 自带版本）：`pip install` 把 pyspark 装进解释器 site-packages 解决 PyCharm 远程解释器静态索引红线 + 本地 pytest；集群运行时同一解释器的 site-packages pyspark 直接命中，HMS 走 `SPARK_CONF_DIR=/etc/spark/conf/hive-site.xml`（见 `01-运行环境.md §4`）
- `findspark` 已于 2026-04-21 随 §三 瘦身删除（冒烟确认其在 CDH + pip pyspark 情境下只做同版本同路径覆盖，行为无差异，详见 `92-进度.md` changelog）

### 7.2 日志改进

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

### 7.2.1 日志路径约定

日志文件统一落 `${LOG_ROOT_DIR}/{module}/{dt}/{file}.log`：

- `LOG_ROOT_DIR` 默认 `${HOME}/log`，外配在 `conf/env.sh`
- `{module}` 取值 `datax` / `spark` / `ds` / `csv` / `export` 等顶层入口名
- `{dt}` 格式 `yyyymmdd`

**待建**：`log_path(module, dt, file)` 工具函数 Python / Shell 各一份，避免入口脚本各自拼。

### 7.3 部署改进

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

### 7.4 DataX 限速逻辑

`job_config_generator.py` 中根据时间段（7:50-19:00 白天限速，其余时间放开）动态设置 DataX 传输速率：
```python
if 750 < local_time < 1900:
    speed = self.get_speed(10, byte=10485760, record=40000)   # 白天低速
else:
    speed = self.get_speed()                                   # 夜间高速
```
**建议**：将时间段和速率配置化，避免硬编码。

### 7.5 Spark / HMS 侧 Ranger Hive 策略验证（低优先级）

**背景**：`02-权限与账号.md §1` 链路 B 原先假设 HMS 侧也挂载 Ranger Hive Plugin，从而 PySpark 直连 HMS 的库/表/列操作仍受 Ranger Hive 管辖。但在 CDH 6.3.2 默认部署中，Ranger Hive Plugin **只挂载在 HiveServer2**，HMS 服务端未必部署；若 HMS 未挂，PySpark / spark-submit 的 SQL 层授权会绕过 Ranger Hive，**仅剩 NameNode 上的 Ranger HDFS Plugin 做数据平面兜底** —— 与同版本 Impala 的"无 Ranger 细粒度授权，只能 HDFS 兜底"规律一致。

**代码侧查证已完成**：仓库内 `dw_base/spark/spark_sql.py:167` 等 8 处 `.enableHiveSupport()`，Spark 确实走 HMS 读元数据，不走 HS2，因此是否受 Ranger Hive 管辖完全看集群侧 HMS 是否挂插件。

**集群侧验证动作（待做）**：

1. 登 HiveMetastore 节点，查 `hive-site.xml`：有无 `hive.metastore.pre.event.listeners = org.apache.ranger.authorization.hive.authorizer.RangerHiveMetastoreAuthorizer`
2. 查 HMS 进程 classpath 里是否包含 `ranger-hive-plugin-*.jar`
3. 从开发环境跑一条故意触发 column 级禁权的 PySpark SQL，看是否被拒 —— 端到端佐证
4. 结论回填到 `02-权限与账号.md §1` 链路 B 关键点段 + 关键点段里的 mermaid Note

**后续处理**：若 HMS 未挂 Ranger Hive，调研补挂成本 + 评估现有 HDFS 兜底是否足够（大部分数仓读写场景下足够，因为 PySpark 任务绝大多数以受控 Unix 账号提交、权限粒度粗即可；若要满足敏感列屏蔽类需求则必须补挂）。

## 八、聚簇推进视图

替换原 P0-P3 线性优先级表。按聚簇 A-F 组织（定义见 §〇），同一聚簇内部不强排序，跨聚簇只标强依赖。

### A 配置外移 / 硬编码清理

| 子项 | 状态 | 依赖 | 参见 |
|------|------|------|------|
| `conf/env.sh`（LOG_ROOT_DIR / RELEASE_USER / RELEASE_ROOT_DIR / PYTHON3_PATH / DATAX_HOME） | 待启动 | — | §2.1 / §7.2.1 |
| `conf/workers.ini`（DataX Workers + 权重 map 外移） | ✅ 2026-04-23 | — | §2.1 |
| `conf/alerter.ini`（告警 Webhook，入库） | 待启动 | 旧告警代码删除（已 2026-04-20 完成） | §2.1 |
| `conf/spark-defaults.conf`（底层 11 条）+ `conf/spark-tuning.conf`（调优 10 条）+ `spark_sql.py` 三级覆盖 | ✅ 2026-04-21 | — | §2.3 |
| `conf/datax-speed.ini`（DataX 分时速率） | 待启动 | — | §2.9 |
| datasource 扁平化 `{db_type}/{env}-{实例简称}.ini` + db_type 按父目录段判定 | ✅ 2026-04-22 | — | §2.1 / §2.5 |
| DataX 脚本去前缀剥离 + JSON 输出目录改名扁平化 `conf/datax-json/{ini_basename}.json` | 待启动 | — | §2.1 / §2.5 |

### B `dw_base/` 重组

| 子项 | 状态 | 依赖 | 参见 |
|------|------|------|------|
| **B1** `__init__.py` 瘦身（修剪式） | ✅ 2026-04-21 | — | §三 |
| **B2** `common/utils/io/ops` 四模块边界定稿 | 待启动 | — | §2.10 |
| **B3** `__contains__` → `in` 全局替换 | 待启动 | — | §4.1 |
| **B3** Shell/Python 环境检测去重（`bin/common/init.sh` ↔ `dw_base/__init__.py`） | 待启动 | B1 | §4.2 |
| **B3** `mysql_utils.py` SQL 注入修复 | ✅ 2026-04-21 随 `dw_base/database/` 整目录删自动完成；mysql 读写后续在 `dw_base/io/db/` 下重造 | — | 见 §5.2 |
| **B4** 新占位模块 `io/` / `ops/` / `pm/` / `dq/` / `wiki/` 骨架 | ✅ 2026-04-20 已建（2026-04-21 去掉各模块 README，docstring 即文档） | B2 边界规则先立 §2.10 | §2.11 |

### C `bin/` 入口收口

| 子项 | 状态 | 依赖 | 参见 |
|------|------|------|------|
| `bin/publish.sh`（从项目根挪入） | ✅ 2026-04-20 完成 | — | §2.1 行 |
| `bin/excel_to_hive.py` 删除（有需求重做） | ✅ 2026-04-20 完成 | — | changelog |
| `bin/csv-to-hdfs-starter.py` 实现 | 待启动 | A 大部分就绪 | kb/30 §4.3 |
| `bin/datax-import` + `bin/datax-export` 两命令收口 | 待启动 | **B2** 四模块边界 / A datax 路径解耦 | §2.6 |
| `bin/datax-gc-generator.py` 从零重写 | 待启动 | datasource 多环境就绪 | §2.7 |

### D 基础设施

| 子项 | 状态 | 依赖 | 参见 |
|------|------|------|------|
| `tests/{unit,integration}/` 骨架 + README | ✅ 2026-04-20 完成 | — | §2.11 / §六 |
| `tests/unit/udf/test_spark_common_udf.py` 首批单测 | 待启动 | tests 骨架 + `conftest.py` | §六 |
| DataX 配置生成单测 | 待启动 | — | §六 |
| 告警模块重写（弃钉钉 → `conf/alerter.ini`） | 待启动 | `conf/alerter.ini` | §2.1 / 阶段 4 |
| 日志模块统一（Python `logging` 双输出 / `log_path` 工具） | 待启动 | `conf/env.sh` LOG_ROOT_DIR | §7.2 / §7.2.1 |
| Hive HDFS 小文件合并工具重新实现 | 待启动 | **B2** → `ops/` | 阶段 4 |
| 分区保留工具重新实现（元表驱动 + 参数化） | 待启动 | **B2** → `ops/` | 阶段 4 |
| `dq/` 数据质量规则首批 + runner | 待启动 | `conf/alerter.ini` 就绪 | §2.11 |
| `pm/` TAPD / Jira 集成（commit → 任务 ID） | 待启动 | 确认平台选型 | §2.11 |
| `wiki/` Docmost → `kb/inbox/` | 待启动 | Docmost API 鉴权确认 | §2.11 |
| Ranger Hive 策略验证（集群侧） | 待启动 | — | §7.5 |

### F 老代码删除（阶段 5）

见 `kb/92-重构进度.md` 阶段 5 checklist。前置：新业务 SQL 在生产稳定运行一个完整周期（新业务 SQL 开发不属于重构 scope，详见 kb/92 变更记录）。

### 当前推进建议

**本阶段可并行开工**（无前置阻塞）：

1. A 大部分子项（env.sh / workers.ini / alerter.ini / datax-speed.ini）
2. B2 四模块边界定稿（只需写决策，不改代码）
3. B3 代码风格修正（`__contains__` 全替换）
4. D 首批 UDF 单测（tests 骨架已建）

**等待前置**：

- C `bin/` 两命令 / `datax-gc-generator` 重写 ← B2 + A datax
- D `ops/` 下两个工具 ← B2
- F `launch-pad/` 整删 ← 新业务 SQL 生产稳定一个完整周期（新业务开发本身不属于重构 scope）

## 九、待讨论议题

以下议题已登记为未来专题讨论对象，暂不展开、不纳入聚簇推进：

- **分布式任务分发**：DolphinScheduler worker 分发机制 × 新入口 `dw_base/datax/` 内部实现（worker 选择 / 批量展开 / 日志路径 / ssh 并发）的交互关系。