ETL 基础与工具选型：别在开头就选错路

数据整合听起来高大上，其实就是把数据从 A 搬到 B，再洗一洗、切一切。很多团队一上来就纠结「用 Spark 还是 Flink」，结果半年过去了，连数据都没跑通。我见过太多这样的项目了。

这一章，咱们不绕弯子。直接聊清楚四件事：ETL 和 ELT 到底怎么选？批处理和流处理分别解决什么问题？主流工具哪些值得上手？

ETL 还是 ELT？哪套更适合你

说白了，传统 ETL 是「先做饭再端菜」——数据在进入目标系统之前，先完成清洗、转换、聚合。ELT 则是「先上菜再加工」——数据原封不动地先堆进数据湖，什么时候用什么时候再转换。

我在早期的一个医疗项目里，就吃过 ETL 的亏。当时非要把所有数据都标准化后再加载，结果下游一直变字段格式，ETL 脚本反复重写。后来换成了 ELT，先落湖再通过视图做转换，迭代速度快了两倍不止。

批处理 vs 流处理：不是非此即彼

很多人以为批处理和流处理是互斥的。其实不是。你想想看，大多数场景白天跑流处理做近实时分析，晚上跑批处理做全量校对和回溯。我建议先想清楚你的数据延迟要求：

场景	推荐模式	延迟要求
报表、月度统计	批处理	小时级 / 天级
实时监控、异常检测	流处理	秒级
多组学特征提取	批 + 流混合	分钟级 ~ 天级
基因变异实时分析	流处理为主	秒级 ~ 分钟级

常用工具横向对比

下面这几个工具，基本覆盖了 90% 的数据湖 ETL 场景。我按自己的使用体验排了序：

Apache Spark

Spark 是大数据批处理的事实标准。它的 Structured Streaming 也能做流处理。我用 Spark 做基因组数据清洗特别顺手——一次读取数万个小文件，性能远超 Python 原生方法。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("multiomics_etl").getOrCreate()

# 读取多个表达矩阵文件
df = spark.read.option("header", "true").option("sep", "\t") \
    .csv("s3://omics-landing/expression/*.tsv")

# 过滤低表达基因
df_clean = df.filter(df.expression_value > 0.1)

# 写出到 Parquet
df_clean.write.mode("overwrite").parquet("s3://omics-lake/expression_clean/")

Apache Airflow

Airflow 不是真正的 ETL 引擎，它是任务编排器。但数据管线里的依赖关系、重试、告警，全靠它兜底。

from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime

default_args = {"owner": "omics_team", "retries": 2}

with DAG(
    dag_id="multiomics_pipeline",
    start_date=datetime(2024, 1, 1),
    schedule_interval="0 2 * * *",
    default_args=default_args,
    catchup=False
) as dag:

    extract_rnaseq = SparkSubmitOperator(
        task_id="extract_rnaseq",
        application="/opt/etl/scripts/extract_rnaseq.py",
        conn_id="spark_default"
    )

    extract_proteomics = SparkSubmitOperator(
        task_id="extract_proteomics",
        application="/opt/etl/scripts/extract_proteomics.py",
        conn_id="spark_default"
    )

    [extract_rnaseq, extract_proteomics] >> transform_merge

我记得有一次某个上游数据源的 API 改了格式，Airflow 自动重试了 3 次都没成功，直接给我发短信了。这种兜底能力，在无人值守的半夜管线里特别重要。

dbt（data build tool）

dbt 是 ELT 模式里的明星。它只管「T」——转换。数据已经落在湖里了，你用 SQL 定义模型，dbt 帮你自动管理依赖和增量更新。

-- models/expression_summary.sql
-- 对单细胞表达矩阵做汇总

WITH cell_type_agg AS (
    SELECT
        cell_type,
        avg(log2_expression) AS mean_expression,
        count(DISTINCT gene_id) AS n_genes_detected
    FROM {{ ref('raw_expression') }}
    GROUP BY cell_type
)
SELECT * FROM cell_type_agg

AWS Glue

Glue 是 AWS 的全托管 ETL 服务。如果你整个基础架构都在 AWS 上，它会是性价比很高的选择。Serverless，不用维护集群。

import sys
from awsglue.transforms import *
from awsglue.context import GlueContext
from awsglue.job import Job

glueContext = GlueContext(spark.sparkContext)
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

# 从 Glue Data Catalog 读取
datasource = glueContext.create_dynamic_frame.from_catalog(
    database="omics_metadata",
    table_name="proteomics_raw"
)

# 过滤空值
filtered = Filter.apply(frame=datasource, f=lambda x: x["intensity"] is not None)

# 写出到 Parquet 分区表
glueContext.write_dynamic_frame.from_options(
    frame=filtered,
    connection_type="s3",
    connection_options={"path": "s3://omics-lake/proteomics/"},
    format="parquet"
)
job.commit()

工具选型速查表

工具	适用场景	上手成本	多组学优势
Spark	大规模批处理、ETL	中	支持 Python/Scala/R，生态丰富
Airflow	管线编排、依赖管理	低	适合多步骤组学流程
dbt	ELT 中的转换层	低	模型管理和数据测试强
AWS Glue	AWS 原生全托管	低	无需运维、集成 Data Catalog

我的习惯做法是：Airflow 做编排，Spark 做批处理 ETL，dbt 做湖内转换层。这套组合在大多数多组学项目里都能跑得很好。如果预算宽裕、团队不大，Glue 也是一个不错的选择，尤其适合只用 AWS 的场景。