提数执行方详解

flowchart TD
  IN["pickExecutor(bo) 入口
PickDataManager:122
设置 callbackUrl = dw.site/api/package/callback"]
  IN --> RT{"quotaCode ∈
dataPick.quotaCodes？"}
  RT -->|"是（prod: user_info）"| CHA["路径 A：自建 datapick-service
executeByDataPickService:135
Feign DataPickApi → /execute
同步返回 code=0 即受理
execid = 本地随机 UUID
不进执行队列、不占 project"]
  RT -->|"否（其余数据源）"| CHB["路径 B：Azkaban 大数据平台
executeByBigDataService:237
login → buildParam → /executor
返回真实 execid
进入执行队列、占用 project"]
  CHA -.->|"异步跑完
FTP 上传结果"| CB["统一回调出口
/api/package/callback
PackageManagerApiImpl:140"]
  CHB -.->|"平台异步跑完
FTP 上传结果"| CB

flowchart TD
  PE["pickExecutor(bo, needQueue=true)
PickDataManager:618"]
  PE -->|"BO 已带 project"| EXE["直接执行"]
  PE -->|"project 为空"| AP["acquireProject(flow):1043
读执行 ZSet 取已占用 project
候选 = 池 − 已占用 → 随机取一个"]
  AP -->|"有空闲"| EXE
  AP -->|"全占用"| WQ2["addPickTaskWaitingQueue:965
member=BO的JSON, score=入队时间戳
返回 code=queued"]
  EXE --> R{"提数请求结果"}
  R -->|"code=OK"| AA["addApplyingProject:1141
member=project:packageCode:execid
score=受理时间戳（超时基准）"]
  R -->|"msg 含 running
（flow 已在跑）"| WQ2
  R -->|"failed"| F["返回 failed"]

flowchart TD
  C["taskCycleExecutor
*/5s（:672）"]
  C --> T["executeTaskTimeOutHandle
扫描执行中 ZSet 全部 flow"]
  C --> W["waitingTaskHandle
扫描等待队列全部 flow"]

  T --> T1{"now − score > 2小时？"}
  T1 -->|"是"| T2["查 callout_package_task
非「提取完成」→ pullStatus=4
remark=提数任务执行超时失败
从 ZSet 删除"]
  T1 -->|"score=null"| T3["脏数据直接删除"]

  W --> W1["Redisson 锁
waitingPickTaskExecuteLock:{flow}
tryLock 不等待/租约10s"]
  W1 -->|"拿到锁"| W2["acquireProject(flow)
无空闲则本轮放弃"]
  W2 --> W3["按 score 升序取最早任务
（严格 FIFO）"]
  W3 --> W4["从等待队列删除
递归调 pickExecutor(bo, true)"]
  W4 --> W5{"结果回写"}
  W5 -->|"OK"| W6["pullStatus=2 提取中"]
  W5 -->|"queued"| W7["留队列等下轮"]
  W5 -->|"failed"| W8["pullStatus=4 + 删除"]

flowchart TD
  E["execute(req)
DataPickServiceImpl:121"] --> QB["getQueryBuilder:359
quotaCode 直接作表名，单表无 JOIN"]
  QB --> BR{"quotaCode ==
dataPick.JD.table？"}
  BR -->|"是"| JD["pickDataJD:656
京东撞库文件分支"]
  BR -->|"否"| MAIN["pickData:538 主循环"]

  subgraph SQLBUILD["SQL 构建 MySQLQueryBuilder"]
    S1["遍历 condition JSON
conditionMap: 运算符→字段→值"]
    S1 --> S2["动态字段：t_create_time ≥ -30
→ create_time ≥ '今天-30天'"]
    S1 --> S3["fileurl 条件 → 不进 WHERE
addFile() 收集文件路径"]
    S1 --> S4["packageTag 非空 → 拼子查询
contact_number NOT IN
(select phone from people_tag ...)"]
    S2 --> S5["SELECT selectFields FROM 表
WHERE (c1 AND c2 ...)"]
    S3 --> S5
    S4 --> S5
  end
  QB -.-> SQLBUILD

  MAIN --> CT["countTotalData
SELECT COUNT(*)"]
  CT --> BL["布隆过滤器初始化
容量=总量, 误判率按数量级自适应"]
  BL --> LOOP{"游标分页循环
pageSize=1000
WHERE ... AND id > 上页最大id
ORDER BY id LIMIT 1000"}
  LOOP --> F1["fetchData:888
敏感字段 AES 解密
（手机号/姓名/身份证/地址）"]
  F1 --> F2["dataFilter:793
文件匹配：内存 contains 过滤
（运营商文件值必须命中）"]
  F2 --> F3["deduplicate:864
布隆过滤器按 distinctFields
跨页去重"]
  F3 --> F4{"超过 limitCount？"}
  F4 -->|"是"| F5["截断并结束"]
  F4 -->|"否"| LOOP
  LOOP -->|"取完"| CSV["writeToCSV:928
清洗引号/换行, 英文逗号→中文
本地文件 = UUID.csv"]
  CSV --> UP["FTP 上传
dataExtractPath 目录"]
  UP --> CB["HTTP POST 回调
callbackUrl（原请求全字段回传
+ success/count/remark/ftpPath）"]
  F5 --> CSV

flowchart TD
  L["1. login()
POST bigdata.url/ 表单
username/password/action=login
→ session.id"] --> P["2. buildParam:479
conditionType=normal → getNormalParam
sql/scala/user_domain → condition 原文透传"]
  P --> C1["命令文本拼装 buildQuery:544
Equal → name isIn MTIz（Base64）
GreaterThan → age gt MTM=
Between → pt_day between xxx
Like → containsArray / NotLike → notcontainsarray"]
  C1 --> X["3. extra_param 组装
command + callback + select + header
+ exclude1（固定剔除）/exclude2（自定义排除）
Map → JSON → UTF-8 → Base64 → URLEncoder"]
  X --> H["4. POST /executor 表单
session.id & ajax=executeFlow
& project={project} & flow={quotaCode}
& flowOverride[extra_param]={base64}"]
  H --> E2["5. 响应 execid
空或 -1 → failed
成功 → 记入执行 ZSet
project:packageCode:execid"]
  E2 -.->|"平台异步跑完"| CB2["回调 /api/package/callback
ftpPath 指向大数据侧上传的提数目录"]