血緣分析流程
Airflow 原生 CLI 與 REST API 服務，Airflow Task 血緣方式可以通過以下幾個步驟獲取：
-
通過 CLI: airflow list_dags 方式獲取 dag 列表list_dags
-
遍歷 dag 列表命满，通過 CLI: airflow show_dag ${dag_id} 方式獲取 dag 描述信息长捧，標紅出即為 dag 內(nèi) task 依賴關系show_dags
-
由于依賴關系存在 ExternalTaskSensor领跛，所以需要通過 REST API 獲取起 Task Info 信息rest
通過綁定關系解析核蘸，生成 Parent - Child 關系即可
核心代碼實現(xiàn)
List Dags
String command = "airflow list_dags";
String result = CommandUtils.exeCommand(command);
// 解析返回數(shù)據(jù)類型
List<String> shellResultLines = CommandUtils.getShellResultLines(result);
int lastIndex = 0;
for(int i=0; i<shellResultLines.size(); i++) {
if (shellResultLines.get(i).contains("--------")) {
lastIndex = i;
}
}
List<String> dagList = shellResultLines.subList(lastIndex + 1, shellResultLines.size());
return dagList;
Show Dag
String command = "airflow show_dag %s";
// check 有效 dag 列表
List<String> validDagList = dagList.stream().map(dag -> {
String showDagCommand = String.format(command, dag);
try {
int exeCode = CommandUtils.exeCommand(showDagCommand, null);
if (exeCode == 0) {
return dag;
}
} catch (IOException e) {
log.error("exe command: {} is error, Exception:{} ", showDagCommand);
}
return null;
}).filter(dag -> dag != null).collect(Collectors.toList());
// 獲取有效 dag 信息
List<Map<String, String>> dagDetailList = validDagList.stream().map(dag -> {
String result = CommandUtils.exeCommand(String.format(command, dag));
HashMap<String, String> dagDetailMap = new HashMap();
dagDetailMap.put(dag, result);
return dagDetailMap;
}).collect(Collectors.toList());
Task
由于部分單獨包含單個 Task 的 Dag，在 Show Dag 無法獲取渺蒿，所以添加無依賴關系 Task
// 提取 task 關系
dagDetailList.stream().forEach(dagDetailMap -> {
List<DagTaskDependencyDO> dagTaskDependencyList = new ArrayList<>();
dagDetailMap.forEach((dag, detail) -> {
// 拿到 dag 下 所有 tasks
List<String> taskList = listTasks(dag);
List<String> shellResultLines = CommandUtils.getShellResultLines(detail);
Set<String> relationTaskSet = new HashSet<>();
shellResultLines.forEach(line -> {
// 檢查包含的依賴關系
if (line.contains("->")) {
String[] taskIds = line.split("->");
String parentTask = taskIds[0].trim().replaceAll("\"|\'", "");
String task = taskIds[1].trim().replaceAll("\"|\'", "");
DagTaskDependencyDO dagTaskDependencyDO = new DagTaskDependencyDO();
dagTaskDependencyDO.setDagId(dag);
dagTaskDependencyDO.setTaskId(task);
dagTaskDependencyDO.setParentDagId(dag);
dagTaskDependencyDO.setParentTaskId(parentTask);
dagTaskDependencyList.add(dagTaskDependencyDO);
relationTaskSet.add(task);
relationTaskSet.add(parentTask);
}
});
// 添加無依賴關系的 task
taskList.removeAll(relationTaskSet);
taskList.forEach(task -> {
DagTaskDependencyDO dagTaskDependencyDO = new DagTaskDependencyDO();
dagTaskDependencyDO.setDagId(dag);
dagTaskDependencyDO.setTaskId(task);
dagTaskDependencyDO.setParentDagId("-");
dagTaskDependencyDO.setParentTaskId("-");
dagTaskDependencyList.add(dagTaskDependencyDO);
});
});
External Sensor
- 部分 External Sensor 對應的 Dag Task痢士，需要替換映射回 External Dag && Task
- ExternalTaskMarker 需要特殊處理，詳見下述代碼
List<DagTaskDependencyDO> suitableDagDetailList = dagTaskDependencyList.stream().map(dagTaskDependency -> {
DagTaskDependencyDO target = new DagTaskDependencyDO();
try {
BeanUtils.copyProperties(target, dagTaskDependency);
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (InvocationTargetException e) {
e.printStackTrace();
}
String response = HttpUtils.getRequest(RestApiHelper.getTaskInfoUrl(env, dagTaskDependency.getDagId(), dagTaskDependency.getTaskId()));
Map<String, String> externalMap = wideExternal(response);
if (!externalMap.isEmpty()) {
if (! target.getParentDagId().equals("-")) {
target.setRawDagId(dagTaskDependency.getDagId());
target.setRawTaskId(dagTaskDependency.getTaskId());
short isExternalSensor = 1;
target.setIsExternalSensor(isExternalSensor);
target.setDagId(externalMap.get(RestResponseConstant.EXTERNAL_DAG_ID));
target.setTaskId(externalMap.get(RestResponseConstant.EXTERNAL_TASK_ID));
}
// 如果存在單個 task，且配置 ExternalTaskMarker怠蹂，則保存 parentDagId && parentTaskId
if (target.getParentDagId().equals("-")) {
short isExternalSensor = 1;
target.setIsExternalSensor(isExternalSensor);
target.setParentDagId(externalMap.get(RestResponseConstant.EXTERNAL_DAG_ID));
target.setParentTaskId(externalMap.get(RestResponseConstant.EXTERNAL_TASK_ID));
}
}
String parentResponse = HttpUtils.getRequest(RestApiHelper.getTaskInfoUrl(env, dagTaskDependency.getParentDagId(), dagTaskDependency.getParentTaskId()));
Map<String, String> parentExternalMap = wideExternal(parentResponse);
if (!parentExternalMap.isEmpty()) {
target.setRawParentDagId(dagTaskDependency.getParentDagId());
target.setRawParentTaskId(dagTaskDependency.getParentTaskId());
short isExternalSensor = 1;
target.setIsExternalSensor(isExternalSensor);
target.setParentDagId(parentExternalMap.get(RestResponseConstant.EXTERNAL_DAG_ID));
target.setParentTaskId(parentExternalMap.get(RestResponseConstant.EXTERNAL_TASK_ID));
}
return target;
}).collect(Collectors.toList());
batchInsert(env, suitableDagDetailList);
});
結果
