任务简介¶

限制¶

• Serverless Task 的最大计算大小相当于 XXLARGE 虚拟仓库。

• 总任务时间可能包括任务开始前的延迟。有关更多信息，请参阅 任务持续时间。

使用无服务器计算模型创建任务¶

如果使用 CREATE TASK 或 ALTER TASK，请省略 WAREHOUSE 参数。

执行任务的角色必须具有全局 EXECUTE MANAGED TASK 权限。有关任务的访问控制要求的更多信息，请参阅 任务安全性。

您可以通过设置以下参数来控制 Serverless Task 的成本和性能：

• SERVERLESS_TASK_MAX_STATEMENT_SIZE：仓库的最大大小，以防止出现意外费用。

• SERVERLESS_TASK_MIN_STATEMENT_SIZE：仓库的最小大小，以确保性能可预测。

• TARGET_COMPLETION_INTERVAL：任务应执行的时间窗口。

指定这些设置时，优先级如下：

• SERVERLESS_TASK_MAX_STATEMENT_SIZE

• SERVERLESS_TASK_MIN_STATEMENT_SIZE

• TARGET_COMPLETION_INTERVAL

Snowflake 如何估算无服务器计算资源¶

Snowflake 会根据目标时间范围以及之前任务运行的性能自动估算完成任务所需的计算资源。

目标时间范围可以通过 TARGET_COMPLETION_INTERVAL 设置。如果未设置此值，Snowflake 会调整无服务器计算资源的大小，以在下一个计划的执行时间之前完成。

任务完成后，Snowflake 会按以下方式估算所需的计算资源：

• 如果任务完成得过快（在比目标时间范围少一个时间因子的时间内完成），Snowflake 会在后续运行中将资源规模缩小一个 仓库大小。

• 如果任务完成得过慢（在比目标时间框架多一个时间因子的时间内完成），Snowflake 会在后续运行中将资源规模扩大一个 仓库大小。

示例¶

示例 1：创建一个每小时运行一次的 Serverless Task，且目标完成间隔为 120 分钟：

CREATE TASK SCHEDULED_T1 SCHEDULE='USING CRON 0 * * * * America/Los_Angeles'
TARGET_COMPLETION_INTERVAL='120 MINUTE' AS SELECT 1;

from datetime import timedelta
from snowflake.core.task import Cron, Task

tasks = root.databases["TEST_DB"].schemas["TEST_SCHEMA"].tasks

task = tasks.create(
    Task(
        name="SCHEDULED_T1",
        definition="SELECT 1",
        schedule=Cron("0 * * * *", "America/Los_Angeles"),
        target_completion_interval=timedelta(minutes=120),
    ),
)

通过将 TARGET_COMPLETION_INTERVAL 设置为一段相对较长的持续时间，这表明任务执行运行更长时间是可以接受的。Snowflake 会自动调整无服务器计算的大小，以便任务执行在指定的 TARGET_COMPLETION_INTERVAL 内完成。

示例 2：创建一个每小时运行一次的 Serverless Task，且目标完成时间为 10 分钟：

CREATE TASK SCHEDULED_T2 SCHEDULE='USING CRON 0 * * * * UTC'
TARGET_COMPLETION_INTERVAL='10 MINUTE'
SERVERLESS_TASK_MAX_STATEMENT_SIZE='LARGE'
AS SELECT 1;

from datetime import timedelta
from snowflake.core.task import Cron, Task

tasks = root.databases["TEST_DB"].schemas["TEST_SCHEMA"].tasks

task = tasks.create(
    Task(
        name="SCHEDULED_T2",
        definition="SELECT 1",
        schedule=Cron("0 * * * *", "UTC"),
        target_completion_interval=timedelta(minutes=10),
        serverless_task_max_statement_size="LARGE",
    ),
)

通过将 TARGET_COMPLETION_INTERVAL 设置为一段相对较短的持续时间，这表明倾向于快速完成任务。Snowflake 会自动调整无服务器计算的大小，以便任务执行在指定的 TARGET_COMPLETION_INTERVAL 内完成，直到达到 SERVERLESS_TASK_MAX_STATEMENT_SIZE。

示例 3：创建一个 Serverless Task，计划每天在目标完成间隔内的某个时刻执行，并设置最小和最大仓库大小范围：

CREATE TASK SCHEDULED_T3 SCHEDULE='USING CRON 0 0 * * * UTC'
TARGET_COMPLETION_INTERVAL='180 M'
SERVERLESS_TASK_MIN_STATEMENT_SIZE='MEDIUM' SERVERLESS_TASK_MAX_STATEMENT_SIZE='LARGE' AS SELECT 1;

from datetime import timedelta
from snowflake.core.task import Cron, Task

tasks = root.databases["TEST_DB"].schemas["TEST_SCHEMA"].tasks

task = tasks.create(
    Task(
        name="SCHEDULED_T3",
        definition="SELECT 1",
        schedule=Cron("0 0 * * *", "UTC"),
        target_completion_interval=timedelta(minutes=180),
        serverless_task_min_statement_size="MEDIUM",
        serverless_task_max_statement_size="LARGE",
    ),
)

通过此配置，Serverless Task 每天午夜至少在 TARGET_COMPLETION_INTERVAL 为 3 小时的中型仓库上运行一次。Snowflake 会自动调整无服务器计算的大小，以便任务执行在指定的 TARGET_COMPLETION_INTERVAL 内完成，直到达到 SERVERLESS_TASK_MAX_STATEMENT_SIZE。

示例 4：创建一个每天计划一次且无目标完成间隔的 Serverless Task：

CREATE TASK SCHEDULED_T4 SCHEDULE='USING CRON 0 0 * * * UTC'
SERVERLESS_TASK_MAX_STATEMENT_SIZE='LARGE' AS SELECT 1;

from datetime import timedelta
from snowflake.core.task import Cron, Task

tasks = root.databases["TEST_DB"].schemas["TEST_SCHEMA"].tasks

task = tasks.create(
    Task(
        name="SCHEDULED_T4",
        definition="SELECT 1",
        schedule=Cron("0 0 * * *", "UTC"),
        serverless_task_max_statement_size="LARGE",
    ),
)

通过此配置，Serverless Task 每天午夜运行一次。如果一系列任务执行持续时间未在下一个计划时间之前完成，Snowflake 会自动调整无服务器计算的大小，以便任务执行在指定的 SCHEDULE 内完成，直到达到 SERVERLESS_TASK_MAX_STATEMENT_SIZE。

有关参数和语法的更多信息，请参阅 CREATE TASK 和 ALTER TASK。

选择仓库大小¶

如果选择使用现有仓库为单项任务提供计算资源，请遵循 仓库注意事项 中描述的最佳实践。要了解任务的计算需求，可根据仓库大小和群集情况，使用不同的仓库分析单项任务或 任务图 的平均运行时间。还应考虑仓库是否由多个进程共享。有关更多信息，请参阅 任务持续时间。

任务计划与夏令时¶

任务定义中的 cron 表达式支持指定时区。当从标准时间转换为夏令时或从夏令时转换为标准时间时，计划的任务可能会出现意外行为。

例如：

• 在从夏令时转换为标准时间期间，计划于美国/洛杉矶时区 (0 1 * * * America/Los_Angeles) 1 AM 开始的任务将运行两次。一次是在1 AM 运行，另一次是在当地时间 1:59:59 AM 变为1:00:00 AM 时运行。

• 在从标准时间转换为夏令时期间，计划于美国/洛杉矶时区 (0 2 * * * America/Los_Angeles) 2 AM 开始的任务将无法运行，因为当地时间会从 1:59:59 AM 变为 3:00:00 AM。

为避免因夏令时而意外执行任务，请考虑以下事项：

• 不要将任务安排在 1 AM 到 3 AM 之间运行。

• 每年两次手动调整在 1 AM 到 3 AM 之间计划任务的 cron 表达式，以补偿时间变化。

• 使用不适用于夏令时的时间格式，如 UTC。

创建触发任务¶

使用 CREATE TASK，并使用 WHEN 子句定义目标流。（请勿包括 SCHEDULE 参数。）

通过选择以下其中一项，为任务分配计算资源：

• 要手动管理资源（用户管理的仓库任务），请添加 WAREHOUSE 参数。

• 要允许 Snowflake 管理资源 (Serverless Task)，请添加 TARGET_COMPLETION_INTERVAL 参数。Snowflake 会估算所需的资源并进行调整，以在此时间内完成任务。

将现有任务从计划任务迁移到触发任务¶

暂停任务，取消设置 SCHEDULE 参数，然后恢复任务。现有任务必须具有在 WHEN 子句中定义的目标流。

ALTER TASK task SUSPEND;
ALTER TASK task UNSET SCHEDULE;
ALTER TASK task RESUME;

将现有的用户管理的触发任务迁移到无服务器触发任务¶

暂停任务。移除 WAREHOUSE 参数并添加 TARGET_COMPLETION_INTERVAL 参数。恢复任务。有关更多信息，请参阅 无服务器任务。

触发任务注意事项¶

• 在 SHOW TASKS 和 DESC TASK 输出中，触发任务的 SCHEDULE 属性显示为 NULL。

• 在 information_schema 和 account_usage 架构的 task_history 视图的输出中，SCHEDULED_FROM 列显示为 TRIGGER。

• 处理多个数据流时，可以使用以下条件参数来管理任务的启动时间：WHEN … AND 和 WHEN … OR。

触发任务限制¶

• 默认情况下，触发任务最多每 30 秒运行一次。您可以修改 USER_TASK_MINIMUM_TRIGGER_INTERVAL_IN_SECONDS 参数，以便更频繁地运行任务，最短间隔为 10 秒。

• 不支持数据共享、目录表、外部表和混合表上的流。

• 如果通过 视图上的流 触发任务，则通过视图上的流查询引用的表的任何更改也将触发该任务，且不受该查询中任何联接、聚合或筛选操作的影响。

触发任务示例¶

示例 1：创建一个用户管理任务，在两个流中任意一个流的数据发生变化时运行：

CREATE TASK my_task
  WHEN SYSTEM$STREAM_HAS_DATA('my_return_stream')
  OR   SYSTEM$STREAM_HAS_DATA('my_order_stream')
  WAREHOUSE = my_warehouse
  AS
    INSERT INTO customer_activity
    SELECT customer_id, return_total, return_date, ‘return’
    FROM my_return_stream
    UNION ALL
    SELECT customer_id, order_total, order_date, ‘order’
    FROM my_order_stream;

示例 2：创建一个任务，当在两个不同的数据流中检测到数据变化时运行。由于任务使用 AND 条件，如果两个流中只有一个有新数据，则会跳过该任务。

由于 TARGET_COMPLETION_INTERVAL 为 120 分钟，Snowflake 会估算并调整在此时间内完成任务所需的计算资源。

CREATE TASK triggeredTask
  WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
  TARGET_COMPLETION_INTERVAL='120 MINUTES'
  AS
    INSERT INTO completed_promotions
    SELECT order_id, order_total, order_time, promotion_id
    FROM orders_stream
    WHERE promotion_id IS NOT NULL;

创建任务¶

要创建任务，角色必须至少具有以下权限：

运行任务¶

创建任务后，任务所有者必须具有以下权限才能运行任务：

此外，该角色必须具有运行任务执行的 SQL 语句所需的权限。

查看任务历史记录¶

要查看任务，您必须具有以下一个或多个权限：

• ACCOUNTADMIN 角色

• 任务的 OWNERSHIP 权限

• 全局 MONITOR EXECUTION 权限

恢复或暂停任务¶

除任务所有者外，具有任务 OPERATE 权限的角色还可以暂停或恢复任务。此角色必须具有包含任务的数据库和架构的 USAGE 权限。不需要其他任何权限。

在恢复任务时，Snowflake 会验证任务所有者角色是否具有 运行任务 中所列出的权限。

创建自定义角色以创建任务¶

Snowflake 需要不同的权限来创建无服务器任务和由用户管理的任务。

例如，要创建由用户管理的任务，可以创建一个名为 warehouse_task_creation 的自定义角色，并就该角色可在其中创建任务的仓库授予角色 CREATE TASK 和 USAGE 权限。

USE SYSADMIN;

CREATE ROLE warehouse_task_creation
  COMMENT = 'This role can create user-managed tasks.';

from snowflake.core.role import Role

root.session.use_role("SYSADMIN")

my_role = Role(
    name="warehouse_task_creation",
    comment="This role can create user-managed tasks."
)
root.roles.create(my_role)

USE ACCOUNTADMIN;

GRANT CREATE TASK
  ON SCHEMA schema1
  TO ROLE warehouse_task_creation;

from snowflake.core.role import Securable

root.session.use_role("ACCOUNTADMIN")

root.roles['warehouse_task_creation'].grant_privileges(
    privileges=["CREATE TASK"], securable_type="schema", securable=Securable(name='schema1')
)

GRANT USAGE
  ON WAREHOUSE warehouse1
  TO ROLE warehouse_task_creation;

from snowflake.core.role import Securable

root.roles['warehouse_task_creation'].grant_privileges(
    privileges=["USAGE"], securable_type="warehouse", securable=Securable(name='warehouse1')
)

作为可以创建 Serverless Task 的角色的示例；创建一个名为 serverless_task_creation 的自定义角色，并授予角色 CREATE TASK 权限和账户级别 EXECUTE MANAGED TASK 权限。

USE SYSADMIN;

CREATE ROLE serverless_task_creation
  COMMENT = 'This role can create serverless tasks.';

from snowflake.core.role import Role

root.session.use_role("SYSADMIN")

my_role = Role(
    name="serverless_task_creation",
    comment="This role can create serverless tasks."
)
root.roles.create(my_role)

USE ACCOUNTADMIN;

GRANT CREATE TASK
  ON SCHEMA schema1
  TO ROLE serverless_task_creation;

from snowflake.core.role import Securable

root.session.use_role("ACCOUNTADMIN")

root.roles['serverless_task_creation'].grant_privileges(
    privileges=["CREATE TASK"], securable_type="schema", securable=Securable(name='schema1')
)

GRANT EXECUTE MANAGED TASK ON ACCOUNT
  TO ROLE serverless_task_creation;

root.roles['serverless_task_creation'].grant_privileges(
    privileges=["EXECUTE MANAGED TASK"], securable_type="account"
)

创建自定义角色来管理任务¶

创建一个具有 EXECUTE TASK 权限的自定义角色，并为此自定义角色授予任何任务所有者角色，以允许更改他们自己的任务。要移除任务所有者角色执行任务的能力，可以从任务所有者角色中撤消该自定义角色。

例如，创建一个名为 taskadmin 的自定义角色，并授予该角色 EXECUTE TASK 权限。将 taskadmin 角色分配给名为 myrole 的任务所有者角色：

USE ROLE securityadmin;

CREATE ROLE taskadmin;

from snowflake.core.role import Role

root.session.use_role("securityadmin")

root.roles.create(Role(name="taskadmin"))

在将账户级权限授予新角色之前，将活动角色设置为 ACCOUNTADMIN

USE ROLE accountadmin;

GRANT EXECUTE TASK, EXECUTE MANAGED TASK ON ACCOUNT TO ROLE taskadmin;

root.session.use_role("accountadmin")

root.roles['taskadmin'].grant_privileges(
    privileges=["EXECUTE TASK", "EXECUTE MANAGED TASK"], securable_type="account"
)

将活动角色设置为 SECURITYADMIN 以表示该角色可将一个角色授予给另一个角色

USE ROLE securityadmin;

GRANT ROLE taskadmin TO ROLE myrole;

from snowflake.core.role import Securable

root.session.use_role("securityadmin")

root.roles['myrole'].grant_role(role_type="ROLE", role=Securable(name='taskadmin'))

有关创建自定义角色和角色层次结构的更多信息，请参阅 配置访问控制。