雪花设置任务依赖关系

Question

我有一个任务，需要在成功完成不同的前置任务后执行。

Say for example below three tasks triggers at same time and calls different stored proc.

CREATE TASK myschema.mytask_1
    WAREHOUSE = mywh
    schedule='USING CRON 0 8 * * MON America/Los_Angeles'
    as call myschema.MY_PROC_1();

CREATE TASK myschema.mytask_2
    WAREHOUSE = mywh
    schedule='USING CRON 0 8 * * MON America/Los_Angeles'
    as call myschema.MY_PROC_2();


CREATE TASK myschema.mytask_3
    WAREHOUSE = mywh
    schedule='USING CRON 0 8 * * MON America/Los_Angeles'
    as call myschema.MY_PROC_3();

然而，我希望下面的第四个任务在上面三个任务都成功完成后执行。如果他们中的任何一个失败了，第四次就不应该触发。简而言之，第四个任务取决于以上三个任务的完成。我通读了一些雪花文档，发现只有一个任务可以设置为依赖项。现在可以通过一个接一个的方式来考虑下面的内容。此外，我也不确定如何评估前一个任务的成功完成，以便继续进行下去。有没有人能帮我以更好的方式实现这一点。这方面的任何帮助都是非常感谢的。

CREATE TASK myschema.mytask_1
    WAREHOUSE = mywh
    schedule='USING CRON 0 8 * * MON America/Los_Angeles'
    as call myschema.MY_PROC_1();

CREATE TASK myschema.mytask_2
    WAREHOUSE = mywh
    AFTER myschema.mytask_1
    as call myschema.MY_PROC_2();


CREATE TASK myschema.mytask_3
    WAREHOUSE = mywh
    AFTER myschema.mytask_2
    as call myschema.MY_PROC_3();
    
CREATE TASK myschema.mytask_4
    WAREHOUSE = mywh
    AFTER myschema.mytask_3
    as call myschema.MY_PROC_4();

Answer 1

如果您连续运行任务，那么任何失败都将停止其余任务的执行，当下一次调度执行到来时，它将从头开始并再次执行。根据您的逻辑，这可能不是您正在寻找的行为。

关于第一个选项，这里的一个可能的解决方案是利用streams来启动第四个任务。这有点非正统，但你可以让它工作。以下是尝试的基本步骤：

1. 在SP成功完成后，3个并行任务中的每一个都需要将一条记录插入到单独的表中，因此这必须是SP中的最后一步。
2. 这3个表中的每个表都需要创建一个流对象。
3. 您将计划任务每分钟运行一次，并使用类似于以下代码的WHEN子句。
4. 在对流执行某些DML语句的第4个任务之后，您将需要执行一些其他任务，以便重置流。

<代码>G29

步骤3示例：

CREATE OR REPLACE TASK mytask_4
  WAREHOUSE = xxxx
  SCHEDULE = '1 MINUTE'
   WHEN SYSTEM$STREAM_HAS_DATA('mytask_1_stream') = True
    AND SYSTEM$STREAM_HAS_DATA('mytask_2_stream') = True
    AND SYSTEM$STREAM_HAS_DATA('mytask_3_stream') = True;

步骤4示例：

CREATE OR REPLACE TASK mytask_5
  WAREHOUSE = xxxx
  AFTER myschema.mytask_4
INSERT * INTO log_table FROM mytask_1_stream;

CREATE OR REPLACE TASK mytask_6
  WAREHOUSE = xxxx
  AFTER myschema.mytask_4
INSERT * INTO log_table FROM mytask_2_stream;

CREATE OR REPLACE TASK mytask_7
  WAREHOUSE = xxxx
  AFTER myschema.mytask_4
INSERT * INTO log_table FROM mytask_3_stream;

还有一点，在这种情况下，mytask_4永远不会使用任何计算，除非所有3个流都包含数据，这意味着您之前的所有3个任务都已成功执行。

Answer 2