📣新登場！TiDB Cloud BYOC、TiDB Cloud Filesystem、PostgreSQL-compatible TiDB Cloud Starter がパブリックプレビュー中。このページは自動翻訳されたものです。原文はこちらからご覧ください。
AI
開発者
ベストプラクティス
API

CREATE TASK

CREATE TASK ステートメントは、指定した SQL ステートメントをスケジュールベースまたは DAG ベースのタスクグラフで実行する新しいタスクを定義するために使用されます。

NOTICE: この機能は、TiDB Cloud Lake でのみ追加設定なしですぐに利用できます。

構文

CREATE [ OR REPLACE ] TASK [ IF NOT EXISTS ] <name>
 WAREHOUSE = <string>
 SCHEDULE = { <num> MINUTE | <num> SECOND | USING CRON <expr> <time_zone> }
 [ AFTER <string>
 [ WHEN <boolean_expr> ]
 [ SUSPEND_TASK_AFTER_NUM_FAILURES = <num> ]
 [ ERROR_INTEGRATION = <string> ]
 [ COMMENT = '<string_literal>' ]
 [ <session_parameter> = <value> [ , <session_parameter> = <value> ... ] ]
AS
{ <sql_statement>
| BEGIN
    <sql_statement>;
    [ <sql_statement>; ... ]
  END;
}

複数の SQL ステートメントは BEGIN ... END; ブロックで囲むことで、タスクがそれらをスクリプトとして順番に実行します。

パラメータ説明
IF NOT EXISTS任意。指定した場合、同じ名前のタスクがまだ存在しないときにのみタスクが作成されます。
nameタスク名です。これは必須フィールドです。
WAREHOUSE必須。タスクで使用する仮想 Warehouse を指定します。
SCHEDULE必須。タスクの実行スケジュールを定義します。分単位、またはタイムゾーン付きの CRON 式で指定できます。
SUSPEND_TASK_AFTER_NUM_FAILURES任意。連続して失敗した回数がこの値に達すると、タスクは自動的に一時停止されます。
AFTERこのタスクの開始前に完了している必要があるタスクの一覧です。
WHEN boolean_exprタスクを実行するために true である必要がある条件です。
ERROR_INTEGRATION任意。特定の タスクエラーのペイロード を適用したタスクエラー通知に使用する通知統合の名前です。
COMMENT任意。タスクのコメントまたは説明として使用される文字列リテラルです。
session_parameter任意。タスク実行時に使用するセッションパラメータを指定します。セッションパラメータは、CREATE TASK ステートメント内で他のすべてのタスクパラメータの後に配置する必要があります。
sqlタスクが実行する SQL ステートメントです。単一のステートメント、または BEGIN ... END; で囲まれたスクリプトを指定できます。これは必須フィールドです。

使用上の注意

  • 単独タスク、またはタスク DAG のルートタスクにはスケジュールを定義する必要があります。そうしない場合、タスクは EXECUTE TASK を使用して手動実行したときにのみ実行されます。

  • DAG 内の子タスクにはスケジュールを指定できません。

  • タスク作成後、タスク定義で指定したパラメータに基づいて実行されるようにするには、ALTER TASK … RESUME を実行する必要があります。

  • WHEN 条件では <boolean_expression> の一部のみがサポートされます。

    タスクの WHEN 句では、以下がサポートされます。

    • STREAM_STATUS は SQL 式内での評価に対応しています。この関数は、指定したストリームに変更追跡データが含まれているかどうかを示します。この関数を使用すると、現在の実行を開始する前に、指定したストリームに変更データが含まれているかどうかを評価できます。結果が FALSE の場合、タスクは実行されません。
    • AND、OR、NOT などのブール演算子。
    • 数値型、文字列型、ブール型の間のキャスト。
    • 等しい、等しくない、より大きい、より小さいなどの比較演算子。

  • 単一のテーブルストリームから変更データを消費する複数のタスクは、それぞれ異なる差分を取得します。タスクが DML ステートメントを使用してストリーム内の変更データを消費すると、ストリームのオフセットが進みます。その変更データは次のタスクでは消費できなくなります。現時点では、1 つのストリームの変更データを消費するのは 1 つのタスクのみにすることを推奨します。同じテーブルに対して複数のストリームを作成し、それぞれを異なるタスクで消費できます。

  • タスクは各実行ごとにリトライされません。各実行は直列です。各スクリプト SQL は 1 つずつ実行され、並列実行は行われません。これにより、タスク実行の順序と依存関係が維持されます。

  • 間隔ベースのタスクは、固定された間隔のタイミングに厳密に従います。つまり、現在のタスク実行時間が間隔単位を超えた場合、次のタスクは直ちに実行されます。そうでない場合、次のタスクは次の間隔単位がトリガーされるまで待機します。たとえば、1 秒間隔で定義されたタスクで、1 回の実行に 1.5 秒かかる場合、次のタスクは直ちに実行されます。1 回の実行に 0.5 秒しかかからない場合、次のタスクは次の 1 秒間隔の開始まで待機します。

  • セッションパラメータはタスク作成時に指定できますが、後から ALTER TASK ステートメントを使用して変更することもできます。例:

    ALTER TASK simple_task SET
    enable_query_result_cache = 1,
    query_result_cache_min_execute_secs = 5;

cron 式に関する重要な注意事項

  • SCHEDULE パラメータで使用する cron 式には、ちょうど 6 つのフィールドを含める必要があります。
  • 各フィールドは次を表します。
    1. 秒 (0-59)
    2. 分 (0-59)
    3. 時 (0-23)
    4. 日 (1-31)
    5. 月 (1-12 または JAN-DEC)
    6. 曜日 (0-6、0 は日曜日、または SUN-SAT)

cron 式の例

  • Pacific Time の毎日午前 9:00:00:

    • USING CRON '0 0 9 * * *' 'America/Los_Angeles'

  • 毎分:

    • USING CRON '0 * * * * *' 'UTC'
    • これは毎分の開始時にタスクを実行します。

  • 毎時 15 分:

    • USING CRON '0 15 * * * *' 'UTC'
    • これは毎時 15 分にタスクを実行します。

  • 毎週月曜日の 12:00:00 PM:

    • USING CRON '0 0 12 * * 1' 'UTC'
    • これは毎週月曜日の正午にタスクを実行します。

  • 毎月 1 日の午前 0 時:

    • USING CRON '0 0 0 1 * *' 'UTC'
    • これは毎月 1 日の午前 0 時にタスクを実行します。

  • 平日の毎日午前 8:30:00:

    • USING CRON '0 30 8 * * 1-5' 'UTC'
    • これは平日（月曜日から金曜日）の毎日午前 8:30 にタスクを実行します。

使用例

CRON スケジュール

CREATE TASK my_daily_task
 WAREHOUSE = 'compute_wh'
 SCHEDULE = USING CRON '0 0 9 * * *' 'America/Los_Angeles'
 COMMENT = 'Daily summary task'
AS
 INSERT INTO summary_table SELECT * FROM source_table;

この例では、my_daily_task という名前のタスクを作成しています。このタスクは compute_wh Warehouse を使用して、source_table から summary_table にデータを挿入する SQL ステートメントを実行します。タスクは CRON 式 を使用して、Pacific Time の毎日午前 9 時 に実行されるようスケジュールされています。

複数のステートメント

CREATE TASK IF NOT EXISTS nightly_refresh
 WAREHOUSE = 'etl'
 SCHEDULE = USING CRON '0 0 2 * * *' 'UTC'
AS
BEGIN
    DELETE FROM staging.events WHERE event_time < DATEADD(DAY, -1, CURRENT_TIMESTAMP());
    INSERT INTO mart.events SELECT * FROM staging.events;
END;

この例では、nightly_refresh という名前のタスクを作成し、複数のステートメントを含むスクリプトを実行します。スクリプトは BEGIN ... END; で囲まれているため、タスクが実行されるたびに DELETE が INSERT より先に実行されます。

動的 SQL (EXECUTE IMMEDIATE)

CREATE OR REPLACE TASK log_ingestion
  WAREHOUSE = 'default'
  SCHEDULE = 1 MINUTE
AS
EXECUTE IMMEDIATE $$
BEGIN
    LET path := CONCAT('@mylog/', DATE_FORMAT(CURRENT_DATE - INTERVAL 3 DAY, '%m/%d/'));

    LET sql := CONCAT(
        'COPY INTO logs.web_logs FROM ', path,
        ' PATTERN = ''.*[.]gz'' FILE_FORMAT = (type = NDJSON compression = AUTO) MAX_FILES = 10000'
    );

    EXECUTE IMMEDIATE :sql;
END;
$$;

この例では、毎分実行されるタスクを作成しています。このタスクは 3 日前 の stage パス（たとえば @mylog/12/15/）を動的に計算し、COPY INTO ステートメントを組み立てて、EXECUTE IMMEDIATE で実行します。

自動一時停止

CREATE TASK IF NOT EXISTS mytask
 WAREHOUSE = 'system'
 SCHEDULE = 2 MINUTE
 SUSPEND_TASK_AFTER_NUM_FAILURES = 3
AS
 INSERT INTO compaction_test.test VALUES((1));

この例では、mytask という名前のタスクを、まだ存在しない場合に作成します。このタスクは system Warehouse に割り当てられ、2 分ごと に実行されるようスケジュールされています。3 回連続で失敗 すると、自動的に一時停止 されます。このタスクは compaction_test.test テーブルに対して INSERT 操作を実行します。

秒単位のスケジューリング

CREATE TASK IF NOT EXISTS daily_sales_summary
 WAREHOUSE = 'analytics'
 SCHEDULE = 30 SECOND
AS
 SELECT sales_date, SUM(amount) AS daily_total
 FROM sales_data
 GROUP BY sales_date;

この例では、daily_sales_summary という名前のタスクを 秒単位スケジューリング で作成しています。このタスクは 30 SECOND ごと に実行されるようスケジュールされています。タスクは analytics Warehouse を使用し、sales_data テーブルのデータを集計して日次売上サマリーを計算します。

タスクの依存関係

CREATE TASK IF NOT EXISTS process_orders
 WAREHOUSE = 'etl'
 AFTER task1
AS
 INSERT INTO data_warehouse.orders SELECT * FROM staging.orders;

この例では、process_orders という名前のタスクを作成し、task1 が正常に完了した後に実行されるよう定義しています。これは、タスクの Directed Acyclic Graph (DAG) における依存関係を作成するのに役立ちます。このタスクは etl Warehouse を使用し、ステージング領域からデータウェアハウスにデータを転送します。

条件付き実行

CREATE TASK IF NOT EXISTS hourly_data_cleanup
 WAREHOUSE = 'maintenance'
 SCHEDULE = USING CRON '0 0 9 * * *' 'America/Los_Angeles'
 WHEN STREAM_STATUS('db1.change_stream') = TRUE
AS
 DELETE FROM archived_data
 WHERE archived_date < DATEADD(HOUR, -24, CURRENT_TIMESTAMP());

この例では、hourly_data_cleanup という名前のタスクを作成しています。このタスクは maintenance Warehouse を使用し、1 時間ごとに実行されるようスケジュールされています。このタスクは、archived_data テーブルから 24 時間より古いデータを削除します。また、db1.change_stream に変更データが含まれているかどうかを STREAM_STATUS 関数で確認し、条件が満たされた場合のみ実行されます。

エラー統合

CREATE TASK IF NOT EXISTS mytask
 WAREHOUSE = 'mywh'
 SCHEDULE = 30 SECOND
 ERROR_INTEGRATION = 'myerror'
AS
 BEGIN
    BEGIN;
    INSERT INTO mytable(ts) VALUES(CURRENT_TIMESTAMP);
    DELETE FROM mytable WHERE ts < DATEADD(MINUTE, -5, CURRENT_TIMESTAMP());
    COMMIT;
 END;

この例では、mytask という名前のタスクを作成しています。このタスクは mywh Warehouse を使用し、30 秒ごとに実行されるようスケジュールされています。このタスクは、INSERT 文と DELETE 文を含む BEGIN ブロックを実行します。両方の文の実行後に、タスクはトランザクションをコミットします。タスクが失敗すると、myerror という名前の エラー統合 がトリガーされます。

セッションパラメータ

CREATE TASK IF NOT EXISTS cache_enabled_task
 WAREHOUSE = 'analytics'
 SCHEDULE = 5 MINUTE
 COMMENT = 'Task with query result cache enabled'
 enable_query_result_cache = 1,
 query_result_cache_min_execute_secs = 5
AS
 SELECT SUM(amount) AS total_sales
 FROM sales_data
 WHERE transaction_date >= DATEADD(DAY, -7, CURRENT_DATE())
 GROUP BY product_category;

この例では、クエリ結果キャッシュを有効にするセッションパラメータ付きで、cache_enabled_task という名前のタスクを作成しています。このタスクは 5 分ごとに実行されるようスケジュールされ、analytics Warehouse を使用します。セッションパラメータ enable_query_result_cache = 1 と query_result_cache_min_execute_secs = 5 は、他のすべてのタスクパラメータの後に指定されており、実行に少なくとも 5 秒かかるクエリに対してクエリ結果キャッシュを有効にします。これにより、基になるデータが変更されていない場合、同じタスクの後続実行でパフォーマンスを向上できます。

タスク実行履歴を表示する

タスクがいつ、どのように実行されたかを確認するには、TASK_HISTORY() テーブル関数を使用します。

SELECT *
FROM TASK_HISTORY(
  TASK_NAME   => 'daily_sales_summary',
  RESULT_LIMIT => 20
)
ORDER BY scheduled_time DESC;

時間範囲や DAG 内のルートタスク ID によるフィルタリングを含むすべてのオプションについては、TASK HISTORY を参照してください。

このページは役に立ちましたか？