在 Snowflake 中构建数据流水线：新手友好指南

数据流水线是数据工程的核心组成部分。它表示将原始数据从源系统移动到数据仓库的一系列步骤，在那里数据可以被清洗、转换，并为分析做好准备。设计良好的流水线可确保数据从摄取到洞察的流动高效、可靠且安全。

Snowflake 是一款功能强大的云端数据平台，特别适合构建可扩展、灵活且具成本效益的数据流水线。凭借内置的数据加载、转换、自动化与监控功能，Snowflake 让数据团队能够简化复杂工作流，专注于从数据中创造价值。

在本教程中，我们将带您完成在 Snowflake 中构建数据流水线的全过程——从环境搭建到使用 Streams 和 Tasks 等原生工具自动化数据转换。部分内容需要了解某些云服务的集成流程和事件管理系统。对此，我建议查阅您所使用云服务的文档。

如果您刚开始使用 Snowflake，建议先学习我们的Snowflake 基础技能路径，动手学习查询、建模与数据管理。

了解 Snowflake 的架构

Snowflake 独特的架构非常适合支持数据流水线。Snowflake 将计算与存储分离，这意味着您可以分别独立扩展二者。

数据集中存储在云存储层，而计算发生在虚拟仓库中，您可以根据需要启动、停止并调整其大小。

这种架构允许多个用户和进程无冲突地访问同一份数据，非常适合包含摄取、转换和分析的数据流水线工作负载。

它还增加了一层安全性，使数据的更改不会直接影响用户查询该数据的能力。

若需了解更多信息并学习此处未讨论的云层内容，推荐阅读这篇Snowflake 架构深度解读。

Snowflake 流水线的关键组件

在每条流水线中，我们都必须考虑必要的步骤。通常，您会先对数据进行暂存，然后再将其加载到 Snowflake。Snowflake 自身负责数据存储，包括表及其模式（schema），而虚拟仓库则提供执行查询和其他数据操作所需的计算资源。

以下是 Snowflake 数据流水线中这些关键组件的清晰定义：

• Stages（阶段）：用于加载/卸载数据的临时或外部存储位置。
• Virtual Warehouses（虚拟仓库）：专指 Snowflake 提供的计算资源。虚拟仓库本质上是大规模并行处理（MPP）计算集群，由多个虚拟机（VM）组成。Snowflake 负责配置并管理这些资源，以高效处理查询并执行数据转换。
• Databases（数据库）：用于组织数据的逻辑容器。通常存放一组表。
• Schemas（模式）：数据库内用于分组相关表和对象的细分结构。它定义了表之间的关联方式。
• Tables（表）：存放结构化数据的地方。

设置您的 Snowflake 环境

让我们确保您已准备好一个可用的 Snowflake 环境。如果您是 Snowflake 新手，请按以下步骤开始：

创建 Snowflake 账户

如果您还没有 Snowflake 账户，目前提供 30 天免费试用（撰写时），包含 400 美元的免费用量。这应该足够我们测试几条示例流水线了！

1. 访问Snowflake 试用并注册免费账户。

Snowflake 试用版注册表单

2. 出于我们的目的，如果您后续继续使用 Snowflake，为节省信用点成本，我们将选择 Standard 版，而非 Enterprise 版。
3. 选择您偏好的云服务商（AWS、Azure 或 GCP）。本教程中的示例将使用 AWS。
4. 系统自动选择的数据中心应是离您地理位置最近的，这通常能提供最佳体验。
5. 创建账户后，登录Snowflake Web UI。

浏览界面

进入后，您会看到：

• Worksheet（工作表）：运行 SQL 查询的地方。
• Databases（数据库）：查看并管理数据库和模式。
• Warehouses（仓库）：监控并控制计算资源。
• History（历史）：跟踪已运行的查询和任务。

如果您想获得更多使用 Snowflake 的建议，请参阅这门Snowflake 基础课程。

创建必要资源

此步骤将涵盖创建虚拟仓库、数据库和模式。在工作表中运行以下 SQL 来设置您的环境：

-- set the user role appropriately
USE ROLE accountadmin;

-- Create a virtual warehouse
CREATE WAREHOUSE datacamp_pipeline_wh WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND = 60;

-- Set our active working warehouse
USE WAREHOUSE datacamp_pipeline_wh;

-- create our Datacamp Sample Database
CREATE OR REPLACE DATABASE datacamp_sample_db;

-- create the Datacamp (Sample) Raw Schema
CREATE OR REPLACE SCHEMA datacamp_sample_db.raw_schema;技术支持

将数据加载到 Snowflake

正如我们在Snowflake 中的数据摄取一文中所讨论的，Snowflake 同时支持批量（批处理）和连续加载。批量加载旨在一次性将大量数据上传至表中。这通常按计划定期执行，或借助 Snowpipe 的事件驱动管理实现自动化。

连续（流式）加载可通过 Snowpipe Streaming 等方式实现。要摄取数据，我们必须与云服务商创建数据集成（stage），并使用 COPY INTO 进行加载。

创建 S3 stage 与集成

数据摄取是任何流水线的第一步。本步骤将介绍创建存储集成与 stage 的基本操作。为此，我们假设您已使用 Snowflake 的 IAM 用户将 Snowflake 与 AWS 连接完毕。若未完成，请参阅文档中的Snowflake AWS 配置指南。

-- set Role Context
USE ROLE accountadmin;

-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;

-- this creates the storage integration 
-- The STORAGE_AWS_ROLE_ARN will come from the AWS configuration guide steps
-- We will allow access to all locations and block any with sensitive data
CREATE STORAGE INTEGRATION datacamp_aws
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);

-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format 
TYPE = 'CSV' 
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

USE SCHEMA datacamp_sample_db.raw_schema;

--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
  STORAGE_INTEGRATION = datacamp_aws
  URL = 's3://bucket1/path1/'
  FILE_FORMAT = my_csv_format;技术支持

最后，您应已成功创建了与 AWS 的 Snowflake 集成并建立了 stage。若想回顾 AWS 基础，以下文章是不错的AWS 入门。

使用 COPY INTO 的批量加载

COPY INTO 命令是将数据上传至 Snowflake 的常用方法。我们可以使用上传到 Snowflake 的本地文件，或使用云连接。在以下示例中，我们将使用 stage 来复制一个 CSV 文件。

-- Load data using COPY INTO
-- The following will attempt to load all file in the bucket
-- To specify the particular files you can use the FILES or PATTERN settings
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @my_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');技术支持

使用 Snowpipe 自动化加载

当数据到达速度加快时，每次有新文件都手动加载既低效又耗时。相反，我们可以使用 Snowpipe 从 stage 加载文件。然后，将 AWS 的事件通知与我们的 Snowflake IAM 用户结合起来，以便每次触发 Snowpipe。

-- Let’ s first create our Snowpipe
CREATE PIPE my_pipe AS
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @datacamp_s3_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');技术支持

接下来，我们可以使用类似Amazon SQS 的服务，当目标存储桶接收文件时向我们的 IAM 用户发送通知。

最后，我们可以使用以下语句监控 Snowpipe：

SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;技术支持

现在，您无需再为每次 S3 存储桶中有新文件而手动上传数据！这些云端通知会触发 Snowpipe 来加载数据。

使用 Streams 和 Tasks 进行数据转换

Snowflake 的 Streams 和 Tasks 是用于跟踪和转换变更数据的内置工具。Streams 侧重于跟踪表的更改，而 Tasks 用于自动执行 SQL 语句。

Streams 简介

Stream 会捕获表自上次查询以来的变更数据捕获（CDC）信息——即插入、更新和删除。这些通常以数据操作语言（DML）查询的形式出现。这使我们能够查看某张表在两个不同查询时间点之间的更改。

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;技术支持

随后，您可以查询该 stream，仅获取变更：

SELECT * FROM my_stream WHERE METADATA$ACTION = 'INSERT';技术支持

Tasks 简介

Tasks 允许您按计划或事件触发自动执行 SQL。Tasks 可以调用 SQL、JavaScript、Python 等，以便在表和流水线上自动化各类流程。

-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
  WAREHOUSE = datacamp_sample_db 
SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_schema.aws_processed_data
  SELECT * FROM my_stream WHERE location = ‘USA’;

-- This will allow us to test the task
EXECUTE TASK process_changes_task;技术支持

结合使用 Streams 与 Tasks

这是自动化转换流水线的基础。正如上文所示，Tasks 可以使用 Streams 将新数据拉取到另一张表。一般流程可能如下：

• 创建一个 Stream 以监控数据表的变更
• 创建一个 Task，按计划从该 stream 中拉取数据
• 每次拉取数据时，将数据转换为所需的格式
• 将转换后的数据加载到目标表

通过这种方式，我们便构建了数据流水线的部分环节。结合使用 Streams 与 Tasks，我们可以仅对新加载的数据执行转换。如果只使用 Tasks，我们可能每次想要新数据时都得处理整个原始表，这既昂贵又低效，浪费计算资源。

构建端到端数据流水线

让我们来看一个简单的流水线用例：摄取订单数据、跟踪更改，并将其转换为可用于分析的数据。由于前文已分别介绍如何执行每个步骤，本节将展示这些步骤组合在一起时的样子。

步骤 1：创建云存储集成与 stage

首先，我们需要将 Snowflake 生态与云存储连接起来。我们以 AWS 为例。在 Snowflake 中，我们使用 CREATE STORAGE INTEGRATION 与 CREATE STAGE 命令来建立 AWS 与数据库之间的连接。

-- set Role Context
USE ROLE accountadmin;

-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;

-- this creates the storage integration 
CREATE STORAGE INTEGRATION datacamp_aws
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);

-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format 
TYPE = 'CSV' 
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

USE SCHEMA datacamp_sample_db.raw_schema;

--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
  STORAGE_INTEGRATION = datacamp_aws
  URL = 's3://bucket1/path1/'
  FILE_FORMAT = my_csv_format;技术支持

步骤 2：使用 Snowpipe 摄取数据

第二步，我们创建一个 Snowpipe，用于对原始数据表执行 COPY INTO 命令。

-- Let’ s first create our SnowpipeCREATE PIPE my_pipe AS

COPY INTO datacamp_sample_db.raw_schema.aws_first_upload

FROM @datacamp_s3_stage

FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');技术支持

步骤 3：使用 Streams 捕获更改

随后，我们在这张原始数据表上使用 stream 来跟踪更改。

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;技术支持

步骤 4：使用 Tasks 自动化转换

接着，我们创建一个 Task，将新加载的数据转换并加载到表的处理版本中。该任务将按计划触发以检查新数据。如果没有新数据，Task 将不会运行。

-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
  WAREHOUSE = datacamp_sample_db 
SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_schema.aws_processed_data
  SELECT * FROM my_stream WHERE location = ‘USA’;

-- This will allow us to test the task
EXECUTE TASK process_changes_task;技术支持

步骤 5：监控与管理

最后，我们要确保流水线按预期工作。请务必使用 SHOW PIPES 和 SHOW TASKS 命令验证其配置正确并在运行。然后使用 INFORMATION_SCHEMA 的元数据表跟踪数据加载性能与错误。

在 Snowflake 中开发数据流水线的最佳实践

一如既往，我们希望在开发中遵循最佳实践。需要考虑优化性能与成本、确保数据质量与可靠性，以及实施适当的安全与访问控制。

为性能与成本优化

我们的目标应始终是在尽可能低的成本下高效运行流水线。这样可以尽量减少对服务器的影响，并尽可能及时地提供数据。以下是一些建议：

• 对仓库使用自动挂起（auto-suspend）以避免空闲费用。这能通过按时间计费的机制减少计算成本。
• 加载压缩文件并将其批量划分为最佳大小。文件过小会因需为每个文件建立连接而产生过多网络开销；文件过大则可能占用过多带宽并有失败风险。Snowflake 建议增量大小为 100–250 MB。
• 在低使用时段安排转换作业。这样可以避免资源争用。

确保数据质量与可靠性

数据质量与可靠性是每位数据工程师都必须考虑的关键问题。您要确保加载的数据可用且无错误。也要确保当出现问题时，数据流水线不会意外导致数据丢失。以下是一些数据质量管理建议：

• 在摄取前后验证数据。可通过定期检查暂存表并设置告警来实现。
• 使用 streams 和暂存表隔离并修复问题记录。由于只会拉取变更数据，我们可以通过 streams 将问题影响范围最小化。
• 在需要时应用唯一性约束或去重逻辑。这可防止在不允许的地方出现 NULL 数据，或在数据拉取存在重叠时预先去重。

实施安全与访问控制

最后，确保数据拥有适当的访问与安全控制非常重要。有些表可能包含敏感信息，或只应由特定人员修改。请遵循以下建议在 Snowflake 中妥善管理安全与访问：

• 使用角色与授权（roles and grants）来限制访问。
• 使用列级遮蔽或行访问策略来保护敏感数据。
• 轮换存储凭据并监控 stage 的使用情况。

结语

在 Snowflake 中构建数据流水线是一个精简且可扩展的过程，非常适合初学者迈入数据工程领域。凭借直观的架构、强大的摄取选项以及 Streams 和 Tasks 等内置自动化工具，Snowflake 让数据团队能够快速将原始数据转化为可操作的洞察。

按照本指南，您已具备搭建自有 Snowflake 流水线的基础知识——在同一平台内处理摄取、转换与编排。有关 Snowflake 的更多信息，请参阅以下指南：

• Snowflake 入门
• Snowflake 入门课程
• 如何学习 Snowflake
• Snowflake 中的数据建模
• 2025 年最佳 Snowflake 认证