datastash处理流程.md · hacker/datastash

### 支持类型

- 批次；文件/数据库表
- 准实时批次：准实施小文件，定时转钟，清单转钟
- 实时文件：目录下面的所有文件
- 流数据：Socket，Kafka

### 逻辑概念
具有两级任务的架构，unit（单元），task（分片/混编），大部分的情况下是1对1，在混编或分片场景下1个unit对多个task（task是不同的引擎产生或分片产生）

* 单元任务（unit）：一个完整任务的综合聚合，一个DAG描述
* 子任务（task）：逻辑的执行单元，一个unit可以包含多个task，拆分的原则为一类引擎（engine）拆分在一起，任务分片的拆分，使用调度的依赖串接这些task，完成整个unit的作业，分片场景是多个并发依赖关系

### 处理流程

1. 选主broker，主broker加载所有job配置（10分钟运行一次），批次单元进入批次队列 unit_list常驻、unit_wait_queue 任务队列；
2. Broker从unit_wait_queue中争抢任务，锁定任务在执行中，判断依赖是否到达，未到达更新unit_list的状态为等待，会重新加入排队任务；（需要判断cronTab与依赖条件？）；如果依赖到达，调用WorkDispatcher做任务的分配；长时间依赖判断中的状态需要更新为失败重新进入依赖判断队列。
3. workDispather，依赖“线性规划”算法计算出每个broker需要执行的sharding，拆分任务（按照资源分拆，不一定本broker执行），把sharding任务写入每个broker的workpending队列。
4. 正常broker一直在监听这个队列等待sharding任务的派发；需要监听长时间没有处理workpending或者整个broker挂掉的任务置为失败，并且重新分配另外的broker的workpending去执行。
5. 监控的监听器监控所有分片完成后，完成任务状态的更新。

### ZK数据结构

做为维护心跳与选主的服务使用

- Broker选主
- Broker机器列表与心跳维护
    * 目录结构 `/datastah/broker/heartbeat/<IP>`
    * 数据格式 `{'MaxNet':128, 'MaxIo':100, 'ip':IP}`

### broker 内存级

broker的VM内存中的静态配置，每10分钟刷新一次

- unit_vo
	* 静态的unit的配置信息
	* 无状态字段（在redis中属于共享状态）
	* 无批次，在rediscover共享内存
	* 处理中的unit禁止刷新
	
- datasouceDefine
	* 数据源的配置信息，一次性加载

### Redis数据结构

主体的数据都在redis中，实现分布式争抢，分布式锁，共享内容结构

- unit_current_batchno_list：
    * hash<key:unitCode,value:currentBatchNo>，
    * 当前最新的批次，会有滚动更新批次
    * 没有批次的unit无需更新

- unit_list：
    * hash<key:unitProcId,value: unitVo>，
    * 满足预期需要做的单元，每10分钟从数据库中构建（unitCode+batchNo） ，需要判断正确的批次，并且更新到unit_wait_queue中。
    * 单元状态每次变更都会同步到日志表；unitFailover错误监听，会更新这个状态。
    * 处理成功，从这个里面移除

- unit_wait_queue_{priority}：
    * list<unitProcId>，
    * 等待判断依赖到达的队列，每10分钟重新加载和排序，
    * 优先级的判断顺序，批次倒叙，优先级别，右边加载，每次扫描后就会降级
    * boker左边取右边入队

- unit_process_list：
    * hash<key:unitProcId,value:UnitProcessVo>
    * 执行完成unit的分片后就应该进入该列表，这个对象的状态与unit_state_list的状态应该不一样，存储的处理过程中的一些细节状态
    * 存储处理单元对象的信息，包含分片task的信息，但是没有分片任务的状态
    * 处理中的任务列表，包含当前批次的执行内容，分片汇总信息；

- unit_task_wait_queue：
    * list<unitTaskVo>
    * 监听队列判断任务是否依赖到达，到达后完成任务启动（线性规划主机去实现任务）
    * 这个队列的任务是需要实时（5秒）判断依赖是否到达，只有在处理的unit才能进入这个表
    * unit-sharding 分片错误监听

- unit_task_state_list:
	* hash<key: unitTaskProcId,value: taskState>
	* 存储每个分片的状态
	* 处理完成后，找到所有相同unitCode+batchNo的更新状态，整个批次完成后清理这个里面的状态
	* 这个队列需要监听，failover处理（5分钟启动，选主执行），遍历队列对比unit_task_heartbeat_list5分没有执行，对比当前时间，如果分片5min没有上报状态，需要kill分片对应的程序进程后重新规划，需要实现SQL加载和文件加载的分片环境处理。
	* 每十分钟选主重新放入unit_task_wait_queue或unit_task_dispatch_queue队列

- unit_task_dispatch_queue:
	* list<unitTaskVo>
	* 等待分配work主机的任务，未能有资源分配重新入队（左）
	* 每十分钟需要从unit_tasks_state_list中比较与回复队列（防止任务在分配时失败）

- unit_task_heartbeat_list:
	* hash<key: unitTaskProcId,value:taskHeartBeatVo>
	* 每10秒上报处理状态，包含处理ip，pid，当前时间

- broker_{brokerId}_work_queue：
    * list<UnitTaskVo>
    * 分片后的每个子任务，每个broker一个，broker的工作队列，有任务在队列中必须执行

- set
  id IP 判断依赖 time

### 监听器

- 监听等待依赖判断队列
- Broker等待执行sharding监听器
- Unit执行状态监听器，failover
- Unit_sharding执行状态监听器
- Broker挂起监听器？

###未完成

* failover处理逻辑
* 线性规划逻辑
* 前台DAG的参数
* 前台DP的能力
* 前台监控
* 分片逻辑优化
* task依赖测试
* 异常情况处理
* 测试用例

hacker/datastash

简介

发行版

贡献者

近期动态

hacker/datastash .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

hacker/datastash