📚 课程目录第3章批量数据加载实战

3. 批量数据加载实战

在 Snowflake 中，批量加载数据的核心命令是 COPY INTO。它把对象存储中的文件按指定格式解析后写入目标表，并自动记录加载历史、跳过已加载文件、支持错误容忍与性能调优。理解 COPY INTO 的关键，是先分清“数据放在哪里”和“用什么格式读”这两件事：前者由 Stage 决定，后者由 FILE FORMAT 决定。

一、三类 Stage 的定位与选择

Snowflake 的 Stage 分为三种：内部 Stage（Internal Stage）、外部 Stage（External Stage） 和 用户 Stage / 表 Stage。内部 Stage 由 Snowflake 托管存储，适合把本地文件通过 PUT 上传后再加载；外部 Stage 指向 S3、Azure Blob 或 GCS 等云存储位置，适合数据已经落在对象存储、需要持续增量加载的场景；表 Stage 与某张表绑定，常用于与该表相关的临时文件暂存。

选择依据可以概括为：数据源在云上且长期存在，优先用外部 Stage；数据在本地或需要 Snowflake 托管，用内部 Stage；只想为某张表临时放文件，用表 Stage。无论哪种 Stage，COPY INTO 的语法结构一致，差别主要在 Stage 的引用方式与访问凭证。

二、从内部 Stage 加载

内部 Stage 的典型流程是“建 Stage → PUT 上传 → COPY INTO”。下面是一个可参考的示例，假设要把本地 CSV 订单文件加载到 orders_raw 表：

-- 1. 创建文件格式
CREATE OR REPLACE FILE FORMAT ff_csv
  TYPE = CSV
  FIELD_DELIMITER = ','
  SKIP_HEADER = 1
  NULL_IF = ('NULL','')
  EMPTY_FIELD_AS_NULL = TRUE;

-- 2. 创建内部 Stage
CREATE OR REPLACE STAGE stg_orders
  FILE_FORMAT = ff_csv;

-- 3. 上传本地文件（在 SnowSQL 中执行）
PUT file:///data/orders_2024*.csv @stg_orders AUTO_COMPRESS=TRUE;

-- 4. 加载到目标表
COPY INTO orders_raw
  FROM @stg_orders
  FILE_FORMAT = (FORMAT_NAME = ff_csv)
  PATTERN = '.*orders_2024.*[.]csv[.]gz'
  ON_ERROR = 'CONTINUE';

这里 PUT 会把本地文件上传到 Stage，AUTO_COMPRESS=TRUE 表示自动 gzip 压缩，减少传输量。COPY INTO 中的 PATTERN 用正则筛选文件，避免误加载无关文件。ON_ERROR = 'CONTINUE' 表示遇到坏行继续加载，并把错误记录到加载历史中，适合对少量脏数据容忍度较高的场景。

三、从外部 Stage 与云存储加载

外部 Stage 需要先建立存储集成（Storage Integration）或直接提供凭证。推荐使用 Storage Integration，避免在 SQL 中明文写密钥。以 S3 为例：

CREATE OR REPLACE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/snowflake-role'
  STORAGE_ALLOWED_LOCATIONS = ('s3://my-bucket/orders/');

CREATE OR REPLACE STAGE stg_s3_orders
  URL = 's3://my-bucket/orders/'
  STORAGE_INTEGRATION = s3_int
  FILE_FORMAT = ff_csv;

COPY INTO orders_raw
  FROM @stg_s3_orders
  FILE_FORMAT = (FORMAT_NAME = ff_csv)
  ON_ERROR = 'ABORT_STATEMENT';

外部 Stage 的加载与内部 Stage 几乎一致，区别在于数据不需要 PUT，COPY INTO 直接读取云存储。若数据是 Parquet、ORC、JSON 等半结构化或列式格式，只需把 FILE FORMAT 换成对应类型，例如 TYPE = PARQUET，并用 MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE 让列名自动匹配。

四、错误处理与加载历史

COPY INTO 的错误处理由 ON_ERROR 控制，常见取值有：ABORT_STATEMENT（默认，遇错终止）、CONTINUE（跳过坏行继续）、SKIP_FILE（跳过整个出错文件）、SKIP_FILE_<n>（错误行超过 n 时跳过文件）。选择哪种，取决于数据质量要求与重跑成本。

加载历史是排查问题的第一入口。每次 COPY INTO 都会在 COPY_HISTORY 中留下记录，包含文件名、加载行数、错误数、状态等。可以用下面的查询快速定位失败文件：

SELECT file_name, status, row_count, error_count,
       first_error_message, last_load_time
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'ORDERS_RAW',
  START_TIME => DATEADD('hour', -24, CURRENT_TIMESTAMP())
))
ORDER BY last_load_time DESC;

如果某文件加载失败，修正数据后重新执行 COPY INTO 即可。Snowflake 默认会跳过已成功加载的文件（基于文件元数据），因此重复执行不会造成重复加载；若确实需要重载，可使用 FORCE = TRUE，但要谨慎，避免数据重复。

五、性能调优要点

批量加载性能主要受文件大小、数量、并行度和仓库规格影响。经验上，单个文件压缩后 100～250 MB 较合适，文件过小会导致大量小文件开销，过大则并行度不足。可以通过 PUT 的 PARALLEL 参数或外部存储侧合并文件来优化。

仓库规格决定并行加载能力。加载大文件集时，适当放大 Warehouse 能显著缩短时间；加载完成后及时缩小，控制成本。此外，VALIDATION_MODE 可以在正式加载前只做校验，提前发现格式问题：

COPY INTO orders_raw
  FROM @stg_s3_orders
  FILE_FORMAT = (FORMAT_NAME = ff_csv)
  VALIDATION_MODE = 'RETURN_ERRORS';

下面这张图概括了从数据源到目标表的完整加载链路，以及错误处理与加载历史在其中的位置。

六、一个可复用的加载检查清单

实际执行批量加载时，可以按下面的顺序逐项确认，减少返工：

文件格式：分隔符、表头、NULL 表示、压缩方式是否与 FILE FORMAT 一致。
Stage 权限：内部 Stage 是否有 PUT 权限；外部 Stage 的 Storage Integration 是否允许该路径。
目标表结构：列数、列顺序、数据类型是否匹配；半结构化数据是否用 VARIANT 或 MATCH_BY_COLUMN_NAME。
错误策略：ON_ERROR 取值是否符合数据质量要求；是否先用 VALIDATION_MODE 预检。
加载后核对：查询 COPY_HISTORY 确认 row_count 与 error_count，再抽样比对目标表数据。

掌握以上内容后，无论是内部 Stage、外部 Stage 还是云存储直读，都可以用同一套 COPY INTO 思路完成加载，并通过加载历史与错误策略把批量数据管道做得稳定、可观测、可重跑。