概述ETL是Extract Transform Load简称抽取-转换-加载。ETL对比ELT传统ETL是Extract-Transform-Load数据在传输过程中完成转换数据源 → 提取 → 临时存储 → 转换 → 目标存储 → 分析 ↑ ↓ └──中间处理层─────┘实现架构​# ETL 管道示例classETLPipeline:defextract(self,source_config):从源系统提取数据# 1. 连接源数据库/API/文件source_dataself.connect_source(source_config)# 2. 增量提取或全量提取ifself.use_incremental:last_extract_timeself.get_last_extract_time()datasource_data.get_changes_since(last_extract_time)else:datasource_data.get_all()# 3. 存储到临时区域通常为文件或中间数据库self.save_to_staging(data)returndatadeftransform(self,raw_data):在加载前进行数据转换transformed_data[]forrecordinraw_data:# 数据清洗cleanedself.clean_data(record)# 数据标准化standardizedself.standardize_format(cleaned)# 业务逻辑转换business_logicself.apply_business_rules(standardized)# 数据聚合/计算aggregatedself.aggregate_metrics(business_logic)transformed_data.append(aggregated)# 数据质量检查self.validate_data_quality(transformed_data)returntransformed_datadefload(self,transformed_data,target_config):将转换后的数据加载到目标系统# 1. 连接目标数据仓库targetself.connect_target(target_config)# 2. 执行加载操作ifself.use_upsert:target.upsert(transformed_data)else:target.insert(transformed_data)# 3. 更新元数据self.update_metadata(target_config)ELT模式Extract-Load-Transform先把原始数据加载到目标系统再做转换数据源 → 提取 → 目标存储 → 转换 → 分析 ↑ ↓ └─在目标系统中转换─┘实现架构使用SQL​-- 1. 提取并直接加载到数据仓库COPY raw_customersFROMs3://data-lake/raw/customers/2024/01/*.parquetCREDENTIALSaws_iam_rolearn:aws:iam::123456789012:role/DataLoadRoleFORMAT PARQUET;-- 2. 在数据仓库中进行转换CREATEORREPLACETABLEtransformed_customersASSELECTcustomer_id,UPPER(TRIM(first_name))asfirst_name,UPPER(TRIM(last_name))aslast_name,CASEWHENemailLIKE%%THENemailELSENULLENDasemail,DATE_TRUNC(month,created_at)assignup_month,-- 计算字段CASEWHENtotal_orders10THENVIPWHENtotal_orders5THENRegularELSENewENDascustomer_segment,-- 数据质量检查CASEWHENfirst_nameISNULLORlast_nameISNULLTHEN1ELSE0ENDasmissing_name_flagFROMraw_customersWHEREdeleted_atISNULL;-- 3. 创建数据模型CREATEORREPLACEVIEWcustomer_analyticsASSELECTsignup_month,customer_segment,COUNT(DISTINCTcustomer_id)ascustomer_count,AVG(total_orders)asavg_orders,SUM(total_revenue)astotal_revenueFROMtransformed_customersGROUPBY1,2;现代数据仓库的计算能力已经很强Snowflake、BigQuery、Redshift都能高效处理TB级数据转换。与其在传输层做复杂转换不如让专业的数据仓库来干专业的事。优势保留完整原始数据支持后续重新分析转换逻辑更灵活可随时调整错误恢复更简单不用重新抽取数据性能更好利用目标系统的计算资源数据处理位置对比维度ETLELT转换位置中间处理层ETL服务器目标系统数据仓库/数据湖数据移动多次移动源→中间→目标一次移动源→目标计算资源依赖ETL服务器资源利用目标系统计算能力网络传输传输原始处理后的数据只传输原始数据可扩展性对比扩展维度ETLELT数据量扩展受ETL服务器限制利用云数据仓库弹性扩展计算扩展垂直扩展为主水平扩展按需扩展并发处理有限并发能力高并发MPP架构成本扩展固定成本较高按使用量付费弹性成本成本结构对比分析成本类型ETLELT基础设施ETL服务器、存储、网络云数据仓库使用费开发成本复杂转换逻辑开发SQL开发相对简单维护成本ETL作业监控、故障恢复数据管道监控、优化扩展成本硬件升级、许可证费用按使用量付费弹性隐性成本数据移动成本、延迟成本计算资源优化成本混合模式与最佳实践ETLELTETLT混合模式​架构设计​数据源 ----→ 轻量ETL ---→ 数据湖 -----→ ELT ---→ 数据仓库 ---→ 分析 ↑ ↓ ↓ ↓ └──基础清洗──┘ └─复杂转换───┘ETL和ELT不是非此即彼的选择而是数据架构演进的不同阶段和模式。现代数据架构往往采用混合模式根据具体场景选择最合适的处理方式。关键决策因素包括​数据特性​ 规模、速度、多样性业务需求​ 实时性、灵活性、准确性技术环境​ 现有系统、云成熟度、团队技能成本考虑​ 投资回报、运维成本、扩展性随着云数据平台能力的不断增强ELT模式正在成为现代数据栈的主流选择。但ETL在特定场景下仍然具有不可替代的价值特别是在处理复杂业务逻辑、敏感数据和实时流处理方面。Pathway官网Python生态开源GitHub62.7K Star1.7K ForkETL框架适用于流处理、实时分析、LLM管道、RAG等场景。技术架构纯Python驱动Rust引擎底层用基于Differential Dataflow的Rust引擎做高性能计算但API完全Pythonic零配置就能上手。无GIL限制无JVM开销单机可处理每秒6万条消息延迟在毫秒级声明式数据流用pw.Table和pw.operators定义数据流图系统自动优化执行计划。同样代码本地开发、CI/CD测试、生产部署完全通用实时增量计算支持窗口、聚合、连接等操作数据一来就处理。只重新计算变化的部分不做全量重算原生LLM集成内置LLM扩展模块提供LLM包装器、解析器、嵌入器、拆分器以及内存中的实时向量索引集成LlamaIndex和LangChain300数据源连接器通过Airbyte连接器可接入Kafka、GDrive、PG、SharePoint等300多种数据源自带监控仪表盘可跟踪每个连接器发送的消息数量和系统延迟。实战基于pip安装pip install -U pathway入门示例importpathwayaspwfrompathway.xpacks.llm.document_storeimportDocumentStore tablepw.io.csv.read(data.csv,modestreaming)# 过滤出活跃用户resulttable.filter(pw.this.statusactive)pw.io.csv.write(result,output.csv)pw.run()# 监听目录变化自动增量更新向量库documentspw.io.fs.read(./data/,formatbinary,with_metadataTrue)document_storeDocumentStore(docsdocuments,embedderembedder)Airbyte官网开源GitHub21.6K Star5.3K Fork的ELT解决方案提供300个现成的连接器覆盖主流数据库、API、文件系统、数据仓库。Connector Builder不需要写代码就能创建自定义连接器。通过可视化界面填写API端点、认证方式、数据格式几分钟就能搞定一个连接器。支持REST API、GraphQL、Webhook等各种接口类型。支持OAuth、API Key、Basic Auth等认证方式。提供现成集成插件支持主流调度工具Airflow、Prefect、Dagster、Kestra。提供完整API接口可通过程序化方式管理连接器、配置同步任务、监控执行状态。社区版支持的企业级特性容错机制支持断点续传、自动重试、错误处理监控告警实时同步状态、数据质量检查、异常通知高可用部署Kubernetes集成、负载均衡、水平扩展安全性数据加密、访问控制、审计日志多种部署选项Docker Compose本地开发和小规模部署Kubernetes生产环境和大规模部署Airbyte Cloud免运维、提供更高级功能、商业版SaaS服务社区还提供Helm Chart、Terraform模块等基础设施代码。dltData Load Tool简称面向数据装载和轻量ETL的开源GitHub)5.6K Star562 Fork库用生成器、资源和管道描述数据来源dlt负责schema推断、规范化、加载和状态管理。可用于将API、文件或业务对象落库到DuckDB、PG、BigQuery。官方文档。特点声明资源用dlt.resource描述数据来源提取逻辑和加载逻辑分开自动建模嵌套对象可自动规范化为表减少手写建表和展开字段的工作目标灵活同一套提取代码可切换DuckDB、数据库或云数仓等目标适合API/文件数据入仓、轻量ELT、嵌套JSON规范化、本地DuckDB原型到云数仓迁移。不适合复杂低延迟流处理也不适合已经有Airflow、Flink、Spark Streaming完整托管的重型链路直接替换。实战基于pip安装pip install dlt使用特定目标端按文档安装对应扩展或驱动。importosimporttempfilefrompathlibimportPathimportdlt os.environ[DLT_TELEMETRY]Falseos.environ[RUNTIME__LOG_LEVEL]ERRORworkPath(tempfile.mkdtemp(prefixdlt_demo_))os.chdir(work)dlt.resource(nameorders,write_dispositionreplace)deforders():yield{order_id:1,city:Shanghai,amount:120}yield{order_id:2,city:Beijing,amount:90}yield{order_id:3,city:Shanghai,amount:180}pipelinedlt.pipeline(pipeline_namedaily_python_dlt_demo,destinationduckdb,dataset_namedemo_sales,)infopipeline.run(orders())print(loads:,len(info.loads_ids))print(tables:,, .join(sorted(t[name]fortinpipeline.default_schema.data_tables())))withpipeline.sql_client()asclient:rowsclient.execute_sql(select city, sum(amount) as total from orders group by city order by total desc)forcity,totalinrows:print(f{city}:{total})dlt.resource(namecustomers,write_dispositionreplace)defcustomers():yield{customer_id:1,name:Ada,orders:[{order_id:A-1,amount:120},{order_id:A-2,amount:80}]}yield{customer_id:2,name:Lin,orders:[{order_id:B-1,amount:60}]}pipelinedlt.pipeline(pipeline_namedaily_python_dlt_nested,destinationduckdb,dataset_namenested_sales)infopipeline.run(customers())print(normalized tables:)fortableinsorted(t[name]fortinpipeline.default_schema.data_tables()):print(-,table)withpipeline.sql_client()asclient:rowsclient.execute_sql(select count(*) from customers__orders)print(order rows:,rows[0][0])pipelinedlt.pipeline(pipeline_namedaily_python_dlt_append,destinationduckdb,dataset_nameevent_log)defrun_batch(batch_name,rows):dlt.resource(nameevents,write_dispositionappend)defevents():forrowinrows:yieldrow infopipeline.run(events())print(f{batch_name}: loaded{len(rows)}rows, load_ids{len(info.loads_ids)})run_batch(batch_1,[{event_id:1,kind:view},{event_id:2,kind:click}])run_batch(batch_2,[{event_id:3,kind:view}])withpipeline.sql_client()asclient:totalclient.execute_sql(select count(*) from events)[0][0]rowsclient.execute_sql(select kind, count(*) from events group by kind order by kind)print(total rows:,total)forkind,countinrows:print(f{kind}:{count})注意事项固定pipeline_name和dataset_name明确write_disposition是replace、append、merge对schema变化设置告警为目标端凭据和重试策略做单独配置Starlake官网基于Scala语言、开源GitHub209 Star29 ForkETL数据管道编排工具可以通过声明式的配置方法YAML、SQL简化数据处理流程。官方文档。功能特性主流数据源支持包括Apache Spark、Databricks、Apache Kafka、Amazon Redshift、DuckDB、DuckLake、Google BigQuery、Snowflake、PG、MySQL、Oracle、SQL Server、JDBC、本地文件JSON、CSV、Excel、XML、Parquet等零代码数据提取通过YAML配置文件实现全量或者增量数据提取和加载包括自动化的数据质量验证数据隐私安全控制应用行级和列级安全整个不需要编写任何代码低代码数据转换基于SQL和YAML定义转换操作自动化表级和字段级血缘关系。也可以通过Python脚本实现复杂的数据转换自动化流程编排可生成任务的有向无环图DAG通过集成Airflow、Dagster等流程编排工具实现流程自动化数据治理和质量支持每个操作节点的模式约束、规则验证、质量检查等措施确保数据一致性与合规性多引擎与跨引擎支持为不同的任务模型使用不同的存储引擎例如使用原生数据仓库引擎执行简单的加载操作同时使用Spark引擎处理XML文件或者加载过程中的转换操作本地化测试集成提供转换编译器可以将各种SQL实现转换为本地DuckDB语法不需要配置额外的测试环境就可以验证数据加载、数据转换等流程VS Code插件支持Starlake配置语法高亮、模式验证、SQL代码片段、数据管道可视化等功能。实战基于Docker部署dockerrun-itstarlakeai/starlake:latest其他ETL架构中的38个子系统ETL架构师面试题