pg_cron を使用する

VPC環境で利用できます。

Cloud DB for PostgreSQLで pg_cron extensionを提供していますので、以下のご利用ガイドをご参照ください。pg_cronは、PostgreSQL内部で cron式に基づいて SQLタスクを予約実行する extension機能です。

コンソールによる extensionのインストール

Cloud DB For PostgreSQLでは superuser権限を提供していないので、コンソールを通じて superuser権限が必要な extensionをインストールできます。詳細は、以下のガイドをご参照ください。

• Extensionのインストール

pg_cronをインストールする際は、タスクを予約する対象のデータベースを選択します。インストールプロセス中には DBサーバが再起動され、HA(高可用性)構成では Failover方式によりサーバが順次再起動されます。

pg_cron とは

pg_cronをインストールすると、対象のデータベースに cron スキーマが作成され、以下の構成要素を通じてタスクを予約・管理します。Cloud DB for PostgreSQLでは superuser権限が提供されないため、以下の表の DB ownerアカウント使用の項目に従い、一部の関数は使用できません。

予約されたタスクは、そのタスクを登録したユーザーの権限で実行されます。登録済みのユーザーがアクセスできないテーブルに対しては、タスクを行うことはできません。

タスクの予約と管理

インストール時に選択したデータベースに DB ownerアカウントでアクセスし、タスクを予約します。

タスク予約

cron.schedule(job_name, schedule, command) 形式で予約します。

-- 毎日03:00に sales.ordersテーブルの VACUUM ANALYZEを実行
SELECT cron.schedule('nightly-vacuum-orders', '0 3 * * *', 'VACUUM ANALYZE sales.orders');

-- 毎週日曜日05:00に、90日以上前の注文を削除
SELECT cron.schedule('purge-old-orders', '0 5 * * 0', $$DELETE FROM sales.orders WHERE ordered_at < now() - interval '90 days'$$);

タスクの照会と変更

予約されたタスクは、cron.jobで照会します。スケジュールやコマンドを変更する際は cron.alter_jobを使用できないため、cron.unscheduleで削除してから同じ名前で再度予約します。

-- 予約されたタスクを照会(自分が登録したタスクのみ表示)
SELECT jobid, jobname, schedule, command, active FROM cron.job;

-- スケジュール変更: 既存のタスクを削除した後、新しいスケジュールで再度予約します。
SELECT cron.unschedule('purge-old-orders');
SELECT cron.schedule('purge-old-orders', '0 6 * * 0', $$DELETE FROM sales.orders WHERE ordered_at < now() - interval '90 days'$$);

-- タスク削除
SELECT cron.unschedule('nightly-vacuum-orders');

ユースケース: 日別売上集計の取り込み

毎日早朝に前日の注文を集計し、要約テーブルに取り込むユースケースです。以下のユースケースは、スキーマやテーブルの作成から検証まで、そのまま実行できます。このユースケースでは、pg_cronがインストールされたデータベースを maindb、当該データベースの DB ownerを mainownerと仮定します。すべてのタスクは、maindbに mainownerアカウントでアクセスしてから登録します。

-- maindb に mainownerアカウントでアクセスします。
\c maindb mainowner

-- 1.使用するスキーマを作成します(Cloud DB for PostgreSQLでは、publicスキーマの使用は推奨されていないため、別のスキーマを使用します)。
CREATE SCHEMA sales;

-- 2.ソース注文テーブルを作成し、サンプルデータを入力します(運用中のテーブルがある場合は省略します)。
CREATE TABLE sales.orders (
    order_id   bigserial     PRIMARY KEY,
    amount     numeric(12,2) NOT NULL,
    ordered_at timestamptz   NOT NULL DEFAULT now()
);

INSERT INTO sales.orders (amount, ordered_at) VALUES
    (15000, current_date - 1 + time '09:30'),  -- 昨日の注文
    (32000, current_date - 1 + time '13:10'),  -- 昨日の注文
    (8900, current_date - 1 + time '20:05'),  -- 昨日の注文
    (47000, current_date     + time '08:00');  -- 今日の注文(昨日の集計には含まれない)

-- 3.集計結果を保存するための要約テーブルを作成します。
CREATE TABLE sales.daily_sales_summary (
    sales_date   date          PRIMARY KEY,
    order_cnt    integer       NOT NULL,
    total_amount numeric(14,2) NOT NULL
);

-- 4.まず、動作を迅速に検証するために、毎分実行するようにタスクを予約します。
--    スキーマを含む名前を使用し、同じ日付で再実行されても重複しないよう、ON CONFLICTで更新します。
SELECT cron.schedule(
    'rollup-daily-sales',
    '* * * * *',
    $$INSERT INTO sales.daily_sales_summary (sales_date, order_cnt, total_amount)
      SELECT current_date - 1, count(*), coalesce(sum(amount), 0)
      FROM sales.orders
      WHERE ordered_at >= current_date - 1
        AND ordered_at <  current_date
      ON CONFLICT (sales_date)
      DO UPDATE SET order_cnt    = EXCLUDED.order_cnt,
                    total_amount = EXCLUDED.total_amount$$
);

-- 5.1~2分ほど待ってから、取り込み結果とタスク実行履歴を確認します。
SELECT sales_date, order_cnt, total_amount
FROM sales.daily_sales_summary
ORDER BY sales_date DESC
LIMIT 10;

SELECT jobid, status, return_message, start_time
FROM cron.job_run_details
WHERE jobid = (SELECT jobid FROM cron.job WHERE jobname = 'rollup-daily-sales')
ORDER BY start_time DESC
LIMIT 5;

-- 6.正常に動作することを確認したら、テストタスクを削除して運用スケジュール(毎日00:05)として再度予約します。
--    (cron.alter_jobは使用できないため、unscheduleを行った後、再度予約します)。
SELECT cron.unschedule('rollup-daily-sales');
SELECT cron.schedule(
    'rollup-daily-sales',
    '5 0 * * *',
    $$INSERT INTO sales.daily_sales_summary (sales_date, order_cnt, total_amount)
      SELECT current_date - 1, count(*), coalesce(sum(amount), 0)
      FROM sales.orders
      WHERE ordered_at >= current_date - 1
        AND ordered_at <  current_date
      ON CONFLICT (sales_date)
      DO UPDATE SET order_cnt    = EXCLUDED.order_cnt,
                    total_amount = EXCLUDED.total_amount$$
);

実行履歴のモニタリング

cron.job_run_detailsで、タスクの実行履歴を確認できます。自分が登録したタスクの履歴のみ照会されます。

-- 最近の実行履歴を照会
SELECT jobid, jobname, status, return_message, start_time, end_time
FROM cron.job_run_details
ORDER BY start_time DESC
LIMIT 20;

cron.job_run_detailsは自動的にクリーンアップされないため、以下のようにクリーンアップタスクを予約することをお勧めします。

-- 毎週日曜日05:00に、30日以上前の実行履歴を削除
SELECT cron.schedule('cleanup-cron-history', '0 5 * * 0', $$DELETE FROM cron.job_run_details WHERE end_time < now() - interval '30 days'$$);