第二十三讲：云数据仓库ETL——Redshift、BigQuery、Snowflake与云原生工具

说实话，几年前我还在跟传统数仓死磕的时候，压根没想到云数据仓库会发展得这么快。现在你随便问一个做ETL的同行，大概率都在跟这三家打交道：AWS Redshift、Google BigQuery、Snowflake。再加上配套的云原生ETL工具，整个生态已经相当成熟了。

这一讲，我就把这几大云数仓的核心玩法，以及它们对应的ETL工具，给你掰开揉碎了讲清楚。嗯，都是我在实际项目中踩过的坑和总结的经验。

一、三大云数据仓库的ETL差异

先看一张图，帮你快速建立整体认知：

云数据仓库ETL核心架构 数据源（RDS/日志/API） 云原生ETL工具 Redshift BigQuery Snowflake 关键差异： • Redshift：列式存储 + 节点集群，适合大吞吐量批量加载 • BigQuery：无服务器架构，按查询扫描量计费，适合分析型ETL • Snowflake：计算与存储分离，支持多集群并发，弹性伸缩 • 云原生ETL工具：Glue（AWS）、Dataflow（GCP）、Snowpipe（Snowflake）

二、AWS Redshift：传统数仓的云上延续

Redshift给我的感觉，就像把Oracle或Teradata搬到了云上。它基于PostgreSQL 8.x内核改造，但存储引擎完全重写成了列式存储。

2.1 数据加载方式

我个人习惯用COPY命令从S3加载数据。这是最快的方式，没有之一。

-- 从S3加载CSV数据到Redshift
COPY sales_data
FROM 's3://my-bucket/sales/2024/'
IAM_ROLE 'arn:aws:iam::123456789:role/RedshiftS3ReadOnly'
CSV
IGNOREHEADER 1
DELIMITER ','
REGION 'us-east-1';
小技巧： 用COPY命令时，记得把数据文件压缩成gzip格式。我在项目中试过，加载速度能提升3-5倍，而且S3传输费用也省了。

2.2 排序键与分布键

Redshift的ETL性能，很大程度上取决于表设计。说白了，排序键和分布键选对了，查询快得像开挂；选错了，慢到你想砸键盘。

-- 创建表时指定排序键和分布键
CREATE TABLE orders (
    order_id BIGINT,
    customer_id INT,
    order_date DATE,
    amount DECIMAL(10,2)
)
DISTKEY(customer_id)
SORTKEY(order_date);
避坑指南： 我曾经把DISTKEY设成了高基数字段（比如order_id），结果数据分布严重倾斜，一个节点忙死，其他节点闲死。后来改成customer_id，性能直接翻倍。

三、Google BigQuery：无服务器的大数据引擎

BigQuery跟Redshift完全是两种思路。它不需要你管理集群，也不关心节点数量。你只管写SQL，它自动分配计算资源。

3.1 外部表与分区

BigQuery支持直接查询GCS（Google Cloud Storage）上的文件，这叫外部表。我经常用它做数据湖的临时查询。

-- 创建外部表指向GCS上的Parquet文件
CREATE EXTERNAL TABLE project.dataset.external_sales
OPTIONS (
    format = 'PARQUET',
    uris = ['gs://my-bucket/sales/*.parquet']
);

-- 查询外部表
SELECT region, SUM(amount) as total_sales
FROM project.dataset.external_sales
WHERE order_date >= '2024-01-01'
GROUP BY region;

3.2 分区与聚簇

BigQuery按扫描量收费，所以减少扫描数据量就是省钱。分区和聚簇是两大法宝。

-- 创建分区表，按日期分区
CREATE TABLE project.dataset.sales_partitioned
PARTITION BY DATE(order_date)
CLUSTER BY region
AS
SELECT * FROM source_table;

我的经验： 有一次客户跑月度报表，没做分区，每次扫描2TB数据，一个月花了3000多刀。加上分区后，每次只扫50GB，费用降到200刀。你想想看，这差距有多大。

四、Snowflake：计算与存储分离的典范

Snowflake是我个人最喜欢的云数仓。它把计算和存储彻底分开，你可以独立扩缩计算资源，存储则按实际使用量付费。

4.1 Virtual Warehouse与自动挂起

Snowflake的ETL核心是Virtual Warehouse（虚拟仓库）。你可以为不同的ETL任务创建不同大小的仓库，用完自动挂起，不产生额外费用。

-- 创建虚拟仓库，用于ETL任务
CREATE WAREHOUSE etl_wh
WITH
    WAREHOUSE_SIZE = 'MEDIUM'
    AUTO_SUSPEND = 300   -- 5分钟无活动自动挂起
    AUTO_RESUME = TRUE;  -- 有查询时自动恢复

-- 使用仓库执行ETL
USE WAREHOUSE etl_wh;

-- 从外部Stage加载数据
COPY INTO sales_table
FROM @my_stage/sales/
FILE_FORMAT = (TYPE = CSV FIELD_OPTIONALLY_ENCLOSED_BY = '"');

4.2 Time Travel与零拷贝克隆

Snowflake有个很酷的功能叫Time Travel，可以查询过去任意时间点的数据。我在做数据回溯时经常用到。

-- 查询30分钟前的表状态
SELECT * FROM sales_table
AT (TIMESTAMP => DATEADD(minute, -30, CURRENT_TIMESTAMP));

-- 零拷贝克隆，用于测试环境
CREATE TABLE sales_dev CLONE sales_table;
实用技巧： 零拷贝克隆不占用额外存储空间，非常适合做ETL测试。我每次上线新任务前，都会克隆一份生产数据到开发环境跑一遍。

五、云原生ETL工具

光有数仓还不够，你得有工具把数据搬进去。下面这三款工具，我基本每天都在用。

5.1 AWS Glue：无服务器ETL

Glue是AWS的托管ETL服务，底层基于Apache Spark。你不需要管理Spark集群，Glue自动帮你搞定。

# Glue ETL脚本示例（Python）
import sys
from awsglue.transforms import *
from awsglue.context import GlueContext
from pyspark.context import SparkContext

sc = SparkContext()
glueContext = GlueContext(sc)

# 从S3读取数据
datasource = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://my-bucket/input/"]},
    format="json"
)

# 数据转换
transformed = ApplyMapping.apply(
    frame=datasource,
    mappings=[
        ("id", "string", "id", "string"),
        ("price", "double", "amount", "decimal"),
        ("timestamp", "string", "event_time", "timestamp")
    ]
)

# 写入Redshift
glueContext.write_dynamic_frame.from_jdbc_conf(
    frame=transformed,
    catalog_connection="redshift-conn",
    connection_options={
        "dbtable": "public.sales",
        "database": "dev"
    }
)
避坑指南： 我曾经用Glue处理每天上亿条日志，默认的DPU（数据处理单元）数量不够，任务跑了6小时还没完。后来把Worker类型从Standard改成G.1X（内存优化），时间缩短到40分钟。记住，Glue的资源配置很关键。

5.2 Google Dataflow：基于Apache Beam

Dataflow是GCP的流批一体ETL工具。它基于Apache Beam，一套代码可以同时跑批处理和流处理。

// Dataflow批处理示例（Java）
public class BatchETL {
    public static void main(String[] args) {
        Pipeline p = Pipeline.create(
            PipelineOptionsFactory.fromArgs(args).create()
        );
        
        // 从GCS读取
        p.apply("ReadFromGCS", TextIO.read().from("gs://my-bucket/input/*.csv"))
         .apply("ParseCSV", ParDo.of(new DoFn<String, KV<String, Double>>() {
             @ProcessElement
             public void processElement(ProcessContext c) {
                 String[] fields = c.element().split(",");
                 c.output(KV.of(fields[0], Double.parseDouble(fields[1])));
             }
         }))
         .apply("SumByKey", Sum.doublesPerKey())
         .apply("WriteToBigQuery", BigQueryIO.writeTableRows()
             .to("project:dataset.sales_summary")
             .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
        
        p.run();
    }
}

5.3 Snowpipe：Snowflake的自动加载

Snowpipe是Snowflake的自动数据加载服务。你只要把文件放到云存储上，Snowpipe自动检测并加载到表中。

-- 创建Pipe，自动加载S3上的数据
CREATE PIPE my_pipe
AUTO_INGEST = TRUE
AS
COPY INTO sales_table
FROM @my_s3_stage
FILE_FORMAT = (TYPE = CSV);

-- 查看Pipe状态
SELECT SYSTEM$PIPE_STATUS('my_pipe');

六、选型建议

说了这么多，到底怎么选？我根据项目经验给你列个参考：

场景 推荐方案 理由
已有AWS生态 Redshift + Glue 集成度高，S3/RDS等无缝对接
大数据分析为主 BigQuery + Dataflow 无服务器，按量付费，适合Ad-hoc查询
多云或混合云 Snowflake + Snowpipe 跨云支持好，计算存储分离
实时流处理 Dataflow（Beam） 流批一体，延迟低

嗯，云数据仓库的ETL就讲到这里。说白了，没有银弹，只有最适合你业务场景的方案。我建议你每个都上手试一下，跑几个真实任务，感受一下差异。实践出真知嘛。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →