数据分析运行镜像与算子开发（data-analysis runner）

CubeStudio 数据分析任务流（workflow=data_analysis）的运行镜像。源：images/data-analysis/README.md、images/data-analysis/Dockerfile、images/data-analysis/app/。

数据分析调度引擎的整体使用见 数据ETL；调度引擎/任务流框架的二次开发（提交流程、all_template 算子模板、对接外部调度系统等）见 数据ETL调度引擎对接。本篇聚焦运行镜像构建与算子二次开发。

一、用途与运行流程

平台后端在用户点"提交"时（见 myapp/utils/k8s_data_analysis.py）：

1. 把 pipeline 的 dag_json + pipeline.config 写成 K8s ConfigMap。
2. 创建 K8s Job：把 ConfigMap 挂到 /etc/dag/（只读），hostPath 挂到 /pipeline_results/。
3. 启动命令固定为 python /app/runner.py /etc/dag/dag.json。

镜像内 runner.py 解析 DAG，按拓扑序顺序调用每个算子函数，把每步预览写到 /pipeline_results/<pipeline_id>/<run_id>/<task_id>/preview.json （预览目录由 app/ctx.py 构造：/pipeline_results/{pipeline_id}/{run_id} 下再按 task_id 建子目录）。

⚠️ 注意（与旧版源文档不一致，已按代码更正）：当前实现不再引用 K8s Secret。数据源凭据已直接以明文写入 dag_json / ConfigMap，因此容器内没有 /etc/secrets/<name>/ 挂载。如需提升安全性，请收紧该 ConfigMap 的 RBAC。依据：myapp/utils/k8s_data_analysis.py:9,301。

二、构建镜像

镜像默认地址：ccr.ccs.tencentyun.com/cube-studio/data-analysis-runner:20260601，可在 config.py 用 DATA_ANALYSIS_IMAGE 覆盖（后端默认值见 myapp/utils/k8s_data_analysis.py:36 与 myapp/views/view_etl_pipeline_data_analysis.py:1390-1395）。

构建命令（来自 images/data-analysis/Dockerfile 头部注释，多架构构建并推送）：

docker buildx build --platform linux/amd64,linux/arm64 \
  -t ccr.ccs.tencentyun.com/cube-studio/data-analysis-runner:20260601 . --push

说明：images/data-analysis/README.md 中的构建命令仍写作 ./build.sh / PUSH=1 ./build.sh（配 REGISTRY 环境变量），但该目录下并无 build.sh 文件，请以上面 Dockerfile 头部注释中的 docker buildx 命令为准。

镜像基于 python:3.11，Python 依赖见 images/data-analysis/requirements.txt，使用阿里云 PyPI 源。

三、目录结构

app/
├── runner.py              # 主入口：解析 DAG → 拓扑序 → 顺序调用算子 → 写预览
├── ctx.py                 # Context 类（变量替换、preview/progress dump、节点日志）
├── preview.py             # 预览序列化（DataFrame head + dtypes + shape）
└── operators/
    ├── __init__.py        # OPERATOR_REGISTRY = {"csv_read": ingest.csv_read, ...}
    ├── _utils.py          # 算子公共工具
    ├── ingest.py          # 数据接入算子（mysql/postgres/hive 查询、csv/excel/parquet/json 读取、http_api_fetch 等）
    ├── row_ops.py         # 行操作算子（过滤/采样/去重/排序/加索引等）
    ├── col_ops.py         # 列操作算子
    ├── agg_join.py        # 聚合 / 关联算子
    ├── window_ops.py      # 窗口算子
    ├── feature.py         # 特征算子
    ├── ml.py              # 机器学习算子
    ├── control.py         # 流程控制算子
    ├── platform.py        # 平台相关算子
    ├── custom.py          # 自定义算子
    └── result.py          # 结果展示 / 告警算子

（算子分组以代码为准，当前后端 UI 模板共 47 个算子，见 myapp/views/view_etl_pipeline_data_analysis.py:8。）

四、运行环境约定

• 容器内 /etc/dag/dag.json ← ConfigMap 挂载（只读）
• 容器内 /etc/dag/pipeline_config.json ← ConfigMap 挂载（runner.py 在 dag.json 同目录下查找此文件）
• 容器内 /pipeline_results/ ← hostPath 挂载（预览/日志写入）
• 环境变量（runner.py:28-33 读取，均有默认值）：PIPELINE_ID / PIPELINE_NAME / RUN_ID / CREATOR / PROJECT_NAME / NOW_TS

五、算子二次开发

新增一个算子函数分三步：

1. 在 app/operators/<group>.py 里写函数，签名约定为 (ctx, upstream, **params)：

   # 自定义算子：从上游取第一个 DataFrame 并处理后返回
   def my_op(ctx, upstream, **params):
       df = next(iter(upstream.values()))
       # ... 处理 ...
       return df_or_anything
   

2. 在 app/operators/__init__.py 的 OPERATOR_REGISTRY 注册一行映射（key 为算子模板名，value 为函数引用）：

   OPERATOR_REGISTRY = {
       "csv_read": ingest.csv_read,
       "my_op":    mygroup.my_op,   # 新增
       ...
   }
   

   未注册的模板名在运行时会抛错 未知算子: <name>（__init__.py:109-111）。

3. 在 myapp/views/view_etl_pipeline_data_analysis.py 用 _make_template(...) 增加该算子的 UI 模板（参数 schema），并加入算子模板列表，前端画布才能配置该算子参数。

具体以源码为准，仅供导航。