雪花任务-可视化层次结构

Question

雪花允许使用AFTER语法创建一系列任务。

创建任务 后串 指定当前任务的前一个任务。当前面任务的运行成功完成时，它会触发此任务(在短暂滞后之后)。 此参数允许定义一棵简单的任务树；即一组由其依赖项组织的任务。在此上下文中，树是一系列以计划的根任务开始并通过它们的依赖项连接在一起的任务。

假设我们有以下情况：

CREATE DATABASE TEST;
CREATE WAREHOUSE Developer WITH WAREHOUSE_SIZE = 'XSMALL' 
       WAREHOUSE_TYPE = 'STANDARD';
CREATE SCHEMA TEST;

CREATE OR REPLACE TASK task1 WAREHOUSE = Developer SCHEDULE = '10 minute'
AS SELECT system$wait(20);

CREATE OR REPLACE TASK task2 WAREHOUSE = Developer AFTER task1
AS SELECT system$wait(30);

CREATE OR REPLACE TASK task3 WAREHOUSE = Developer AFTER task2
AS SELECT system$wait(60);

CREATE OR REPLACE TASK task4 WAREHOUSE = Developer AFTER task1
AS SELECT system$wait(20);

CREATE OR REPLACE TASK task5 WAREHOUSE = Developer AFTER task1
AS SELECT system$wait(30);

CREATE OR REPLACE TASK task6 WAREHOUSE = Developer AFTER task3
AS SELECT system$wait(40);

CREATE OR REPLACE TASK task7 WAREHOUSE = Developer AFTER task5
AS SELECT system$wait(50);

CREATE OR REPLACE TASK task8 WAREHOUSE = Developer AFTER task5
AS SELECT system$wait(30);

其目标是获得任务的图形表示，以便快速概述或编写文档。

Answer 1

可以使用内置的查看任务历史记录生成图。

任务图以DAG格式描述根任务和子任务的执行。可以使用任务图执行以下操作：

• 显示任务信息，包括根任务和选定子任务的状态。
• 检查从父任务到子任务的任务图。
• 选择任务元素以查看其他详细信息。

输出：

​

​

Answer 2

雪花支持：受抚养人表函数：

此表函数返回简单任务树中给定根(即父)任务的子任务列表。

SELECT CONCAT_WS('.', DATABASE_NAME, SCHEMA_NAME, NAME) AS TASK_NAME, PREDECESSOR
FROM TABLE(INFORMATION_SCHEMA.TASK_DEPENDENTS(TASK_NAME => 'task1',
                                              RECURSIVE => TRUE ));
/*
TASK_NAME        PREDECESSOR
TEST.TEST.TASK1 
TEST.TEST.TASK2 TEST.TEST.TASK1
TEST.TEST.TASK4 TEST.TEST.TASK1
TEST.TEST.TASK5 TEST.TEST.TASK1
TEST.TEST.TASK3 TEST.TEST.TASK2
TEST.TEST.TASK7 TEST.TEST.TASK5
TEST.TEST.TASK8 TEST.TEST.TASK5
TEST.TEST.TASK6 TEST.TEST.TASK3
*/

使用“图作为代码”和美人鱼的思想，我们可以生成以下流程图：

WITH RECURSIVE cte AS (
   SELECT CONCAT_WS('.', DATABASE_NAME, SCHEMA_NAME, NAME) AS TASK_NAME, *
   FROM TABLE(INFORMATION_SCHEMA.TASK_DEPENDENTS(
                   TASK_NAME => 'task1', RECURSIVE => TRUE )) 
                             -- here goes task name
), rec AS (
   SELECT 
      0 AS lvl, cte.TASK_NAME, cte.PREDECESSOR,
      REPLACE(REPLACE(REPLACE(
        'ROOT{.} -- "SCHEDULE: <schedule>;CONDITION: <condition>" --> <root>'
       ,'<schedule>', COALESCE(cte.SCHEDULE, '<none>'))
       ,'<condition>', COALESCE(cte.CONDITION,'<none>'))
       ,'<root>', cte.TASK_NAME)  AS GRAPH_ENTRY
   FROM cte 
   WHERE PREDECESSOR IS NULL
   UNION ALL
   SELECT rec.lvl + 1 AS lvl, cte.TASK_NAME, cte.PREDECESSOR,
          REPLACE(REPLACE('<T1> --> <T2>'
          ,'<T1>', cte.PREDECESSOR)
          ,'<T2>', cte.TASK_NAME) AS GRAPH_ENTRY
   FROM rec
   JOIN cte ON rec.TASK_NAME = cte.PREDECESSOR
)
SELECT 'graph TD' || CHAR(13) || 
       LISTAGG(CHAR(9) || GRAPH_ENTRY || CHAR(13), '') 
               WITHIN GROUP(ORDER BY lvl) AS flow_chart
FROM rec;

我们将得到以下输出：

graph TD
    ROOT{.} -- "SCHEDULE: 10 minute;CONDITION: <none>" --> TEST.TEST.TASK1
    TEST.TEST.TASK1 --> TEST.TEST.TASK2
    TEST.TEST.TASK1 --> TEST.TEST.TASK4
    TEST.TEST.TASK1 --> TEST.TEST.TASK5
    TEST.TEST.TASK2 --> TEST.TEST.TASK3
    TEST.TEST.TASK5 --> TEST.TEST.TASK7
    TEST.TEST.TASK5 --> TEST.TEST.TASK8
    TEST.TEST.TASK3 --> TEST.TEST.TASK6

它可以使用美人鱼-现场编辑可视化。

​

​

美人鱼流程图- LiveDemo

Extras:它还可以用于可视化使用甘特图：执行的历史

启用所有任务：

ALTER TASK TEST.TEST.TASK8 RESUME;
ALTER TASK TEST.TEST.TASK7 RESUME;
ALTER TASK TEST.TEST.TASK6 RESUME;
ALTER TASK TEST.TEST.TASK5 RESUME;
ALTER TASK TEST.TEST.TASK4 RESUME;
ALTER TASK TEST.TEST.TASK3 RESUME;
ALTER TASK TEST.TEST.TASK2 RESUME;
ALTER TASK TEST.TEST.TASK1 RESUME;
SHOW TASKS;

生成甘特图：

SELECT 
  CONCAT_WS('.', DATABASE_NAME, SCHEMA_NAME, NAME) AS TASK_NAME,
  QUERY_START_TIME,
  COMPLETED_TIME,
  DATEDIFF(SECOND, QUERY_START_TIME,  COMPLETED_TIME) AS DURATION_SEC,
  TASK_NAME || ':' || TO_VARCHAR(QUERY_START_TIME, 'YYYY-MM-DD HH:MI:SS') 
            || ',' || DURATION_SEC || 's' AS GRAPH_ENTRY,
  s.gantt || LISTAGG(CHAR(9) || GRAPH_ENTRY || CHAR(13), '') 
         WITHIN GROUP(ORDER BY QUERY_START_TIME) OVER() AS graph
FROM TABLE(information_schema.task_history(
           scheduled_time_range_start=>'2021-05-16 07:00:00.000'::TIMESTAMP_LTZ))
,LATERAL(SELECT REPLACE(
'gantt
    title Task execution
    dateFormat YYYY-MM-DD HH:mm:ss
    axisFormat  %Y-%m-%d %H:%M
    section RunId=<run_id>
'
,'<run_id>'
,RUN_ID)
) s(gantt)
WHERE STATE = 'SUCCEEDED'
  --AND RUN_ID = x
ORDER BY scheduled_time;

输出：

gantt
    title Task execution
    dateFormat YYYY-MM-DD HH:mm:ss
    axisFormat  %Y-%m-%d %H:%M
    section RunId=xxxxxx
    TEST.TEST.TASK1:2021-05-16 07:13:45,20s
    TEST.TEST.TASK5:2021-05-16 07:14:06,31s
    TEST.TEST.TASK4:2021-05-16 07:14:09,21s
    TEST.TEST.TASK2:2021-05-16 07:14:15,30s
    TEST.TEST.TASK8:2021-05-16 07:14:51,34s
    TEST.TEST.TASK7:2021-05-16 07:14:51,50s
    TEST.TEST.TASK3:2021-05-16 07:15:01,60s
    TEST.TEST.TASK6:2021-05-16 07:16:15,40s

​

​

美人鱼甘特图现场演示

增编-支持多个前身：

任务: DAG支持-预览

WITH RECURSIVE cte AS (
   SELECT CONCAT_WS('.', DATABASE_NAME, SCHEMA_NAME, NAME) AS TASK_NAME,
          TRIM(sub.VALUE) AS PREDECESSOR, t.SCHEDULE, t.CONDITION
   FROM TABLE(INFORMATION_SCHEMA.TASK_DEPENDENTS(
                   TASK_NAME => 'task1', RECURSIVE => TRUE )) t
                             -- here goes task name
   ,LATERAL SPLIT_TO_TABLE(COALESCE(PREDECESSOR, ''), ',') sub
), rec AS (
   SELECT 
      0 AS lvl, cte.TASK_NAME, cte.PREDECESSOR,
      REPLACE(REPLACE(REPLACE(
        'ROOT{.} -- "SCHEDULE: <schedule>;CONDITION: <condition>" --> <root>'
       ,'<schedule>', COALESCE(cte.SCHEDULE, '<none>'))
       ,'<condition>', COALESCE(cte.CONDITION,'<none>'))
       ,'<root>', cte.TASK_NAME)  AS GRAPH_ENTRY
   FROM cte 
   WHERE PREDECESSOR = ''
   UNION ALL
   SELECT rec.lvl + 1 AS lvl, cte.TASK_NAME, cte.PREDECESSOR,
          REPLACE(REPLACE('<T1> --> <T2>'
          ,'<T1>', cte.PREDECESSOR)
          ,'<T2>', cte.TASK_NAME) AS GRAPH_ENTRY
   FROM rec
   JOIN cte ON rec.TASK_NAME = cte.PREDECESSOR
)
SELECT 'graph TD' || CHAR(13) || 
       LISTAGG(CHAR(9) || GRAPH_ENTRY || CHAR(13), '') 
               WITHIN GROUP(ORDER BY lvl) AS flow_chart
FROM rec;

示例：

-- extending scenario in question
ALTER TASK task6 ADD AFTER task7;
ALTER TASK task6 ADD AFTER task8;

输出：

​

​