Snowpipe Streaming无法用默认值时，如何获取Snowflake数据摄入时间？

方案1：在数据生产/传输阶段注入摄取时间戳 #

在数据发送至Kinesis前，由生产者（或流处理节点如Lambda、Flink）为每条事件添加ingestion_timestamp字段，值为当前系统时间戳。数据进入Snowflake时直接携带该字段，无需在Snowflake端处理。

示例（Python生产者代码片段）：

import time
import json

# 原始事件数据
raw_event = {
    "creation_timestamp": "2023-11-13 14:30:00",
    "event_data": "user_login"
}

# 添加摄取时间戳
raw_event["ingestion_timestamp"] = time.strftime("%Y-%m-%d %H:%M:%S")

# 发送到Kinesis
# kinesis_client.put_record(StreamName='your-stream', Data=json.dumps(raw_event), PartitionKey='key')

注意：需保证生产者系统时间与Snowflake服务器时间同步，避免时间偏差影响延迟计算准确性。

方案2：通过Staging表+流+任务补全摄取时间 #

利用Snowflake的流（Stream）和任务（Task）组合，在数据落地后补全摄取时间戳：

1. 创建用于接收原始数据的Staging表（无ingestion_timestamp字段）：

CREATE TABLE kinesis_staging (
    creation_timestamp TIMESTAMP_NTZ,
    event_payload VARCHAR
);

2. 为Staging表创建APPEND_ONLY模式的流，捕获新增数据：

CREATE STREAM staging_stream ON TABLE kinesis_staging APPEND_ONLY = TRUE;

3. 创建包含ingestion_timestamp的目标表：

CREATE TABLE final_events (
    creation_timestamp TIMESTAMP_NTZ,
    event_payload VARCHAR,
    ingestion_timestamp TIMESTAMP_NTZ
);

4. 创建定时任务，将流中数据插入目标表时生成摄取时间：

CREATE TASK populate_ingestion_time
WAREHOUSE = your_warehouse_name
SCHEDULE = '30 SECOND' -- 根据延迟需求调整调度间隔
WHEN SYSTEM$STREAM_HAS_DATA('staging_stream')
AS
INSERT INTO final_events (creation_timestamp, event_payload, ingestion_timestamp)
SELECT creation_timestamp, event_payload, CURRENT_TIMESTAMP()
FROM staging_stream;

-- 启动任务
ALTER TASK populate_ingestion_time RESUME;

该方案的额外延迟由任务调度间隔决定，若需接近实时，可将调度频率设为最小支持的10秒（需对应Snowflake版本支持）。

方案3：通过QUERY_HISTORY视图回溯摄取时间 #

Snowpipe Streaming的写入操作会被记录到QUERY_HISTORY视图中，可通过关联数据与查询历史获取摄取时间：

1. 确保Staging表包含可关联的批次标识（如Kinesis的shard_id、sequence_number范围），方便匹配对应的写入查询。
2. 查询QUERY_HISTORY获取写入操作的时间信息：

SELECT
    q.QUERY_ID,
    q.QUERY_START_TIME AS ingestion_start_time,
    q.QUERY_END_TIME AS ingestion_complete_time,
    q.QUERY_TEXT
FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY(
    DATEADD('HOUR', -2, CURRENT_TIMESTAMP()), CURRENT_TIMESTAMP()
)) q
WHERE q.QUERY_TYPE = 'COPY'
AND q.QUERY_TEXT LIKE '%kinesis_staging%';

3. 将查询结果与Staging表数据关联，计算creation_timestamp与摄取时间的差值。
   此方案适合事后延迟分析，而非实时监控场景。

内容的提问来源于stack exchange，提问作者eugen-fried