Snowflake Event Tablesでログ・トレース・監査を一元管理する方法

Event Tablesは、SnowflakeのUDF・ストアドプロシージャ・Streamlitアプリなどから 出力されるログ・トレース・メトリクスイベントを収集するための専用テーブルです。 従来はデバッグログの確認が困難だったSnowflakeの関数実行に対して、 構造化されたイベントデータをSQLで分析できる可観測性（Observability）基盤を提供します。

Event Tableの作成と有効化

-- Event Tableの作成
CREATE OR REPLACE EVENT TABLE observability_db.events.app_events;

-- アカウントレベルでEvent Tableを有効化（ACCOUNTADMIN必須）
ALTER ACCOUNT SET EVENT_TABLE = 'observability_db.events.app_events';

-- データベースレベルでログレベルを設定
ALTER DATABASE my_app_db SET LOG_LEVEL = 'INFO';

-- スキーマレベルでログレベルを設定（より細かい制御）
ALTER SCHEMA my_app_db.analytics SET LOG_LEVEL = 'DEBUG';

-- 特定のUDF/プロシージャだけログレベルを変更
ALTER FUNCTION my_app_db.public.process_data(VARCHAR)
  SET LOG_LEVEL = 'TRACE';

-- トレースレベルの設定
ALTER DATABASE my_app_db SET TRACE_LEVEL = 'ON_EVENT';

LOG_LEVELの階層と継承

LOG_LEVELはアカウント → データベース → スキーマ → オブジェクト（関数/プロシージャ）の順に 上書き可能です。未設定の場合は上位レベルの設定が継承されます。

TRACE_LEVELの設定

UDF/プロシージャからのイベント出力

-- Python UDFからログを出力
CREATE OR REPLACE FUNCTION process_order(order_json VARIANT)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.10'
HANDLER = 'run'
AS $
import logging
logger = logging.getLogger('order_processor')

def run(order_json):
    logger.info(f"Processing order: {order_json['order_id']}")
    try:
        total = sum(item['price'] * item['qty'] for item in order_json['items'])
        logger.info(f"Order total calculated: {total}")
        return {"order_id": order_json['order_id'], "total": total, "status": "success"}
    except Exception as e:
        logger.error(f"Failed to process order: {str(e)}")
        raise
$;

-- JavaScriptプロシージャからログを出力
CREATE OR REPLACE PROCEDURE sync_inventory()
RETURNS VARCHAR
LANGUAGE JAVASCRIPT
EXECUTE AS CALLER
AS $
  snowflake.log("info", "Starting inventory sync");
  var rs = snowflake.execute({sqlText: "SELECT COUNT(*) FROM inventory"});
  rs.next();
  var count = rs.getColumnValue(1);
  snowflake.log("info", "Inventory records: " + count);
  return "Synced " + count + " records";
$;

-- SQLスクリプトからログを出力
CALL SYSTEM$LOG('INFO', 'Data pipeline started');

-- トレースイベントの出力
CALL SYSTEM$ADD_TRACE('pipeline_run', '{"stage": "transform", "rows": 50000}');

Event Tableの主要カラム

Event Tableへのクエリ例

-- 直近1時間のエラーログを取得
SELECT
  timestamp,
  resource_attributes['snow.database.name']::VARCHAR AS db_name,
  resource_attributes['snow.schema.name']::VARCHAR AS schema_name,
  resource_attributes['snow.executable.name']::VARCHAR AS function_name,
  record['severity_text']::VARCHAR AS severity,
  record['body']['text']::VARCHAR AS message
FROM observability_db.events.app_events
WHERE record_type = 'LOG'
  AND record['severity_text']::VARCHAR IN ('ERROR', 'FATAL')
  AND timestamp >= DATEADD(HOUR, -1, CURRENT_TIMESTAMP())
ORDER BY timestamp DESC;

-- 関数別の実行回数とエラー率を集計
SELECT
  resource_attributes['snow.executable.name']::VARCHAR AS function_name,
  COUNT(*) AS total_events,
  COUNT_IF(record['severity_text']::VARCHAR = 'ERROR') AS error_count,
  ROUND(error_count / total_events * 100, 2) AS error_rate_pct
FROM observability_db.events.app_events
WHERE record_type = 'LOG'
  AND timestamp >= DATEADD(DAY, -7, CURRENT_TIMESTAMP())
GROUP BY function_name
ORDER BY error_rate_pct DESC;

-- トレーススパンの実行時間を分析
SELECT
  scope['name']::VARCHAR AS span_name,
  record_attributes,
  DATEDIFF('millisecond',
    record['start_time']::TIMESTAMP_NTZ,
    timestamp
  ) AS duration_ms
FROM observability_db.events.app_events
WHERE record_type = 'SPAN'
ORDER BY duration_ms DESC
LIMIT 20;

運用ベストプラクティス

• 本番環境はINFO、開発環境はDEBUG：LOG_LEVELをデータベース単位で切り替え、本番ではログ量とコストを抑制する
• 長期保存は通常テーブルにコピー：Event Tableの保持期間は短めに設定し、必要なログを定期的に通常テーブルに移行する
• TRACE_LEVELはALWAYSを避ける：全関数呼び出しをスパンとして記録するため大量のイベントが発生する。ON_EVENTで必要箇所のみ記録する
• アラートとの連携：Event Tableに対するAlert（ALTER ALERT）を設定し、エラー率の急増を自動検知する
• 構造化ログを活用：ログメッセージにJSON形式でコンテキスト情報を含め、後からVARIANTパスで検索可能にする

問題で確認

よくある質問