ETL/ELT & Data PipelineETL/ELT & Data Pipelines

Thuộc bộ kiến thức Data Engineer Roadmap.

Tổng quan

Một data pipeline là một chuỗi các bước tự động di chuyển dữ liệu từ một hoặc nhiều nguồn (source) đến một hoặc nhiều đích (destination), đồng thời áp dụng các transformation dọc đường. Đây chính là “hệ thống ống nước” (plumbing) của vòng đời data engineering: ingestion (./03-data-sources-and-ingestion.md) đưa dữ liệu thô về trong tầm tay, storage và data modeling (./06-data-modeling-and-warehousing.md) cho dữ liệu một “ngôi nhà” và một hình dạng cụ thể, nhưng chính pipeline mới là thứ thực sự chạy — theo lịch (schedule) hoặc liên tục — kết nối hai phần đó lại và giữ cho dữ liệu luôn tươi mới. Không có pipeline, một data warehouse chỉ là một tòa nhà trống không có con đường nào dẫn tới.

Câu hỏi thiết kế trung tâm mà mọi pipeline phải trả lời là: transformation xảy ra khi nào, so với việc load dữ liệu? Câu trả lời chia pipeline thành hai họ — ETL và ELT — và lựa chọn thứ tự đó lan tỏa ảnh hưởng đến tooling, chi phí, độ trễ (latency), và ai trong team có thể thực hiện thay đổi. Note này sẽ đi qua sự khác biệt ETL/ELT, pattern Reverse ETL mới hơn khép lại vòng lặp bằng cách đẩy dữ liệu từ warehouse ngược trở lại các công cụ vận hành (operational tools), dbt như tiêu chuẩn thực tế (de facto standard) cho “T” trong ELT, và các nguyên tắc về độ tin cậy (idempotency, checkpointing) — điều phân biệt một pipeline chỉ đơn thuần “chạy được” với một pipeline có thể tin tưởng chạy không giám sát trong nhiều năm.

Kiến thức nền tảng

Data pipeline thực chất là gì

Bóc bỏ hết lớp vỏ thương hiệu, mọi data pipeline đều đang làm cùng ba việc, theo một thứ tự và tần suất nào đó:

1. Extract — kéo dữ liệu ra khỏi hệ thống nguồn (một OLTP database, một SaaS API, một stream sự kiện).
2. Transform — định hình lại, làm sạch, lọc, join, và aggregate dữ liệu đó thành dạng hữu ích cho đích đến của nó.
3. Load — ghi dữ liệu (đã hoặc chưa transform) vào đích (một warehouse, một lake, một ứng dụng khác).

Pipeline là đoạn code và orchestration nối các bước này lại với nhau, khởi chạy bởi một trigger — lịch cron, một file vừa được đặt vào object storage, một job upstream vừa hoàn tất, hoặc một stream liên tục. Điều thay đổi giữa các kiến trúc không phải là ba động từ trên có xảy ra hay không, mà là “T” nằm ở đâu so với “L” — và đó chính xác là sự khác biệt giữa ETL và ELT.

ETL: Extract, Transform, Load

Trong pattern ETL truyền thống, transformation xảy ra ở một lớp xử lý riêng biệt trước khi dữ liệu chạm tới đích:

Source ──▶ Extract ──▶ Transform (external engine) ──▶ Load ──▶ Destination

Về mặt lịch sử, đây không phải là một lựa chọn phong cách — nó bị ép buộc bởi bài toán kinh tế. Data warehouse của thập niên 1990-2000 (Teradata, Oracle on-prem) có compute và storage đắt đỏ, cố định. Bạn không thể đổ dữ liệu thô, trùng lặp, chưa sạch vào warehouse rồi transform sau; mỗi byte và mỗi CPU cycle bên trong warehouse đều khan hiếm và tốn kém. Vì vậy transformation bị đẩy ra các server ETL riêng (Informatica, IBM DataStage, Talend), được sizing độc lập với warehouse, và chỉ kết quả cuối cùng — sạch, đã được model — mới được load vào.

Cái giá của cách tiếp cận này là sự cứng nhắc về kiến trúc: logic transformation nằm trong một tool riêng với engine thực thi riêng, thường ẩn sau một GUI, tách rời khỏi các analyst và engineer thông thạo SQL — những người hiểu rõ business logic nhất. Thay đổi schema hay yêu cầu transformation mới nghĩa là phải đụng vào một hệ thống khác với hệ thống mà dữ liệu cuối cùng sẽ nằm trong, và không có cách nào dễ dàng để xem lại dữ liệu chưa transform một khi bạn đã quyết định không load nó vào.

ELT: Extract, Load, Transform

ELT đảo ngược thứ tự: dữ liệu thô được load vào đích trước, và transformation xảy ra bên trong đích, sử dụng compute của chính nó:

Source ──▶ Extract ──▶ Load (raw) ──▶ Destination ──▶ Transform (in-warehouse SQL)

ELT trở thành pattern thống trị trong thập kỷ qua vì một lý do đơn giản: cloud data warehouse đã khiến compute trở nên rẻ, co giãn (elastic), và tách rời khỏi storage. Snowflake, BigQuery, và Redshift cho phép bạn scale compute lên cho một transformation job rồi scale xuống ngay sau đó, chỉ trả tiền cho những gì đã dùng, và chi phí storage đủ thấp để việc giữ một bản sao đầy đủ dữ liệu thô của mọi bảng nguồn không còn là một khoản chi đáng kể. Khi việc lưu trữ dữ liệu thô với chi phí rẻ và tính toán co giãn trên nó trở nên khả thi, không còn lý do gì để định hình dữ liệu trước khi nó được load — bạn có thể load mọi thứ nguyên bản rồi quyết định cách model nó sau, với đầy đủ SQL và query engine của chính warehouse.

Điều này kéo theo một số hệ quả thực tế:

• Dữ liệu thô được bảo toàn. Vì bản extract chưa qua chỉnh sửa được load vào warehouse trước bất kỳ transformation nào, bạn luôn có một bản sao có thể audit được của “nguồn thực sự nói gì” — điều này khiến việc debug một transformation sai trở nên đơn giản (chỉ cần diff với raw) và khiến việc tái tạo (re-derive) model mới từ lịch sử trở nên khả thi mà không cần extract lại.
• Logic transformation chuyển vào SQL, được thực thi bởi chính warehouse engine, nghĩa là những người vốn đã biết SQL — analytics engineer, analyst — có thể sở hữu (own) transformation mà không cần học ngôn ngữ độc quyền hay GUI của một tool ETL riêng biệt.
• Transformation trở thành phần mềm có thể version-control. File SQL có thể nằm trong git, được code review, test, và CI/CD — đây chính xác là khoảng trống mà dbt lấp đầy (xem bên dưới).

Những tradeoff còn lại là thật, không bị loại bỏ hoàn toàn: dữ liệu thô và dữ liệu đã transform giờ đều được lưu trữ, gây trùng lặp storage (thường là một sự đánh đổi tốt do storage của warehouse rẻ, nhưng không miễn phí); warehouse giờ cần được sizing cho cả ingestion load lẫn transformation compute, và một transformation job chạy tràn (runaway) sẽ tiêu tốn cùng ngân sách compute với mọi thứ khác; và bạn giờ phụ thuộc vào việc warehouse có tooling transformation SQL native mạnh mẽ (window function, incremental materialization, v.v.) thay vì khả năng chuyên biệt của một engine ETL riêng.

Reverse ETL

Nếu tiền đề của ELT là “load dữ liệu thô vào warehouse, rồi transform nó ở đó,” thì Reverse ETL là pattern khép lại vòng lặp: nó lấy dữ liệu đã được transform, đã model, đang nằm trong warehouse và đồng bộ nó ngược ra ngoài các SaaS tool vận hành (operational) mà các đội kinh doanh dùng hàng ngày.

Warehouse (modeled data) ──▶ Reverse ETL tool ──▶ Operational SaaS tool (Salesforce, HubSpot, Zendesk...)

Vấn đề gốc rễ dẫn tới pattern này: một data team tính toán ra thứ gì đó có giá trị trong warehouse — ví dụ, một customer_health_score được xây dựng từ dữ liệu sử dụng sản phẩm, ticket hỗ trợ, và lịch sử billing kết hợp lại — nhưng con số đó vô dụng nếu chỉ nằm trong một bảng mà chỉ analyst mới query. Đội sales sống trong Salesforce, không phải trong một BI tool. Reverse ETL lấy cột customer_health_score đã tính từ một dbt mart và đẩy nó vào một custom field trên Account object của Salesforce, để một sales rep nhìn thấy nó ngay tại nơi họ làm việc.

Pattern này ra đời vì nó biến warehouse thành nguồn sự thật duy nhất (single source of truth) ngay cả với các hệ thống chưa bao giờ được thiết kế để trở thành đích đến “analytics” — thay vì mỗi team tự xây một tích hợp (integration) riêng lẻ để làm giàu Salesforce, Marketo, hay Intercom bằng derived business logic, logic đó được tính toán một lần duy nhất, chính xác, trong warehouse (có dbt test bảo vệ) và phân phối ra mọi tool cần đến nó. Các use case điển hình: đồng bộ điểm lead score vào CRM, đẩy cờ churn-risk vào tool customer success, cập nhật segment email marketing trong nền tảng marketing automation, hoặc đưa một con số lifetime-value đã tính vào tool hỗ trợ để agent ưu tiên khách hàng VIP.

Các tool chuyên dụng trong mảng này — Hightouch, Census, và (với các khả năng gần với Reverse ETL) Segment — kết nối trực tiếp với warehouse, cho phép bạn map một câu SQL query hoặc một dbt model vào các field trong tool đích thông qua connector dựng sẵn, và xử lý lịch đồng bộ, field mapping, và rate limit API của hàng chục SaaS tool đích để engineer không phải tự viết tay code tích hợp API cho từng cái một.

Khái niệm chính

dbt và “T” thời hiện đại

dbt (data build tool) là công cụ đã biến “T” trong ELT từ những script SQL viết tùy tiện (ad hoc) thành một kỷ luật kỹ thuật (engineering discipline). Ý tưởng cốt lõi của nó đơn giản nhưng mạnh mẽ: transformation chỉ là các câu lệnh SELECT, được viết dưới dạng file .sql, và dbt lo việc biến mỗi file thành một table hoặc view trong warehouse, theo đúng thứ tự phụ thuộc (dependency), kèm theo testing và documentation tích hợp sẵn.

Ba thành phần chính giúp điều này hoạt động:

• Models — một file .sql chứa câu lệnh SELECT; dbt compile và chạy nó thành câu lệnh CREATE TABLE/CREATE VIEW/MERGE (materialization có thể cấu hình).
• source() — một tham chiếu tới một raw table mà dbt không tạo ra (thường là các bảng đã được load thô bởi công cụ ingestion ELT của bạn), được khai báo trong một file YAML để dbt biết dữ liệu của pipeline thực sự bắt đầu từ đâu.
• ref() — một tham chiếu tới một dbt model khác, đây là cách các model được xâu chuỗi lại với nhau. Việc dùng ref() thay vì hardcode tên bảng chính là điều cho phép dbt tự động suy ra dependency graph (DAG) và chạy các model theo đúng thứ tự.

Một quy ước phân lớp (layering) phổ biến, được áp dụng rộng rãi, phản ánh ý tưởng “pipeline của các pipeline” (bên dưới): staging → intermediate → marts.

-- models/staging/stg_orders.sql
-- Staging: 1:1 với một raw source table, chỉ rename/cast nhẹ nhàng, không join.
with source as (
    select * from {{ source('ecommerce', 'orders') }}
),
renamed as (
    select
        id as order_id,
        customer_id,
        cast(ordered_at as timestamp) as ordered_at,
        status,
        amount_cents / 100.0 as amount_usd
    from source
)
select * from renamed

-- models/intermediate/int_orders_joined_customers.sql
-- Intermediate: join/business logic mà chưa đạt hình dạng báo cáo cuối cùng.
with orders as (
    select * from {{ ref('stg_orders') }}
),
customers as (
    select * from {{ ref('stg_customers') }}
)
select
    orders.order_id,
    orders.ordered_at,
    orders.amount_usd,
    customers.customer_id,
    customers.signup_date,
    customers.country
from orders
left join customers on orders.customer_id = customers.customer_id

-- models/marts/fct_orders.sql
-- Mart: bảng cuối cùng, hướng tới business, mà analyst và BI tool query trực tiếp.
select
    order_id,
    customer_id,
    country,
    date_trunc('month', ordered_at) as order_month,
    amount_usd
from {{ ref('int_orders_joined_customers') }}
where status = 'completed'

Vì đây là các file thuần túy, chúng trải qua cùng vòng đời như application code: version-control trong git, review qua pull request, test (các test tích hợp sẵn của dbt như unique, not_null, relationships, cộng thêm custom SQL assertion — xem ./16-testing-for-data-pipelines.md), và document với lineage graph tự động sinh ra. Đây chính là cơ chế cụ thể đằng sau khẳng định ở phần trước rằng ELT cho phép “những người biết SQL sở hữu transformation” — dbt là phương tiện hiện thực hóa điều đó.

Idempotency

Một pipeline được gọi là idempotent nếu chạy nó hai lần với cùng đầu vào cho ra cùng kết quả như khi chạy một lần. Nghe có vẻ là một tiểu tiết kỹ thuật, nhưng đây lại là thuộc tính quan trọng nhất để xây dựng những pipeline có thể fail một cách an toàn — và pipeline sẽ fail, dù là do trục trặc mạng, worker hết bộ nhớ, hay API upstream timeout. Nếu một job idempotent, cách khắc phục cho gần như mọi lỗi đều giống nhau: chỉ cần chạy lại (rerun). Nếu không idempotent, mỗi lần lỗi đều đòi hỏi phải thủ công tìm hiểu xem đã có bao nhiêu thiệt hại một phần (partial damage) trước khi có thể retry an toàn.

Sai lầm kinh điển không idempotent là một phép append mù quáng:

-- KHÔNG idempotent: chạy lại sau một lần fail nửa chừng
-- sẽ nhân đôi mọi row đã lọt qua ở lần chạy đầu tiên.
insert into orders_raw (order_id, customer_id, amount_usd, ordered_at)
select order_id, customer_id, amount_usd, ordered_at
from staging_orders_batch;

Nếu job này fail giữa chừng — chẳng hạn kết nối bị đứt sau khi đã insert 6.000 trong tổng số 10.000 row — chạy lại job sẽ insert lại chính 6.000 row đó thêm một lần nữa. Cách khắc phục là làm cho việc ghi (write) trở nên idempotent, hoặc bằng một upsert native của database:

-- Idempotent: upsert của PostgreSQL. Chạy lại sẽ ghi đè
-- các row đã tồn tại thay vì nhân đôi chúng.
insert into orders_raw (order_id, customer_id, amount_usd, ordered_at)
select order_id, customer_id, amount_usd, ordered_at
from staging_orders_batch
on conflict (order_id)
do update set
    customer_id = excluded.customer_id,
    amount_usd = excluded.amount_usd,
    ordered_at = excluded.ordered_at;

hoặc bằng một MERGE của warehouse (pattern mà hầu hết cloud warehouse và incremental materialization của dbt sử dụng bên dưới):

-- Idempotent: MERGE (cú pháp Snowflake/BigQuery/Databricks).
merge into orders_raw as target
using staging_orders_batch as source
on target.order_id = source.order_id
when matched then update set
    customer_id = source.customer_id,
    amount_usd = source.amount_usd,
    ordered_at = source.ordered_at
when not matched then insert (order_id, customer_id, amount_usd, ordered_at)
    values (source.order_id, source.customer_id, source.amount_usd, source.ordered_at);

Cả hai cách viết lại đều dùng một định danh (identifier) ổn định (order_id) làm khóa cho phép write, nên chạy chính xác cùng một batch qua pipeline bao nhiêu lần cũng hội tụ về cùng một trạng thái cuối — đó chính là bản chất của idempotency, và đó là điều khiến “cứ retry task lỗi” trở thành mặc định an toàn trong một orchestrator (./09-workflow-orchestration.md) thay vì một cuộc điều tra thủ công đầy rủi ro.

Checkpointing và watermarking

Idempotency khiến việc chạy lại một job trở nên an toàn; checkpointing (còn gọi là watermarking) khiến việc chạy lại job trở nên rẻ bằng cách cho phép nó tiếp tục từ điểm đã dừng thay vì xử lý lại từ đầu. Một checkpoint đơn giản là một mốc đánh dấu (marker) được lưu trữ bền vững — một timestamp, một ID tăng dần, một Kafka offset — ghi lại pipeline đã tiến được đến đâu một cách thành công.

-- Bảng watermark theo dõi timestamp xử lý thành công gần nhất cho mỗi source.
create table pipeline_watermarks (
    source_name text primary key,
    last_loaded_at timestamp not null
);

-- Câu query extraction chỉ lấy các row mới hơn watermark thành công gần nhất.
select *
from source_orders
where updated_at > (
    select last_loaded_at from pipeline_watermarks where source_name = 'orders'
);

-- Sau khi load thành công, tiến watermark tới timestamp lớn nhất vừa xử lý.
update pipeline_watermarks
set last_loaded_at = (select max(updated_at) from staging_orders_batch)
where source_name = 'orders';

Không có watermark, một lần fail giữa chừng khi extract dữ liệu một năm nghĩa là phải bắt đầu lại toàn bộ năm đó. Có watermark, một job được resume sẽ tiếp tục chính xác từ điểm đã xác nhận thành công cuối cùng. Watermarking và idempotency bổ trợ cho nhau, không trùng lặp: watermarking giới hạn bao nhiêu công việc một lần retry phải làm lại, còn idempotency đảm bảo rằng bất kỳ phần trùng lặp nào bị xử lý lại (ví dụ watermark đã tiến lên một chút trước khi crash) cũng không làm hỏng dữ liệu ở đích.

Các hình dạng pipeline: fan-in, fan-out, và pipeline-of-pipelines

Kiến trúc pipeline trong thực tế hiếm khi là một đường thẳng duy nhất; chúng kết hợp thành một vài hình dạng lặp lại phổ biến.

Fan-in — nhiều nguồn hội tụ vào một đích duy nhất. Đây là hình dạng phổ biến khi xây dựng một warehouse thống nhất: Salesforce, Stripe, và production Postgres database đều đổ vào cùng một schema trong warehouse, sẵn sàng để join.

1. Salesforce

   Stripe

   Postgres
2. Warehouseraw schema

Fan-out — một nguồn duy nhất nuôi (feed) nhiều đích. Một stream sự kiện đơn hàng có thể đổ vào warehouse để phân tích và vào một search index cho storefront và kích hoạt một webhook tới một đối tác vận chuyển (fulfillment), tất cả từ cùng một sự kiện nguồn.

Order events

• Warehouse
• Search index
• Fulfillment webhook

Pipeline-of-pipelines — một chuỗi các pipeline nơi output của giai đoạn này là input của giai đoạn tiếp theo, thường được tổ chức thành các lớp với mức độ tinh chỉnh tăng dần. Đây chính xác là quy ước staging → intermediate → marts mà dbt khuyến khích: công việc của pipeline staging hoàn tất ngay khi stg_orders tồn tại; pipeline intermediate lấy đó làm đầu vào và join; pipeline marts lấy kết quả đó và định hình cho một dashboard. Việc model điều này như một DAG gồm các pipeline nhỏ hơn thay vì một script khổng lồ duy nhất nghĩa là một lỗi hoặc một thay đổi schema ở một lớp bị cô lập và không đòi hỏi phải tái tạo lại các lớp không liên quan, và mỗi lớp có thể được test và sở hữu (own) độc lập.

Các managed ingestion tool và vị trí của chúng

Một phần lớn của “E” (và một phiên bản “L” mỏng, không có nhiều quan điểm riêng) trong các pipeline ELT hiện đại không còn được viết tay nữa — nó được xử lý bởi các managed ingestion tool duy trì sẵn hàng trăm connector dựng sẵn cho các nguồn phổ biến và đổ dữ liệu thô vào warehouse theo lịch, để data engineer dành thời gian cho “T” thay vì viết lại một Salesforce API client lần thứ năm. Các tên tuổi chính:

Tool	Mô hình	Ghi chú
Fivetran	SaaS được quản lý hoàn toàn	Catalog connector lớn, xử lý schema tự động toàn diện, tính phí theo usage
Airbyte	Open-source + tùy chọn managed cloud	Có thể tự host, catalog connector open-source lớn, mở rộng được với connector tùy chỉnh
AWS Glue (ETL jobs)	ETL managed dựa trên Spark trên AWS	Spark serverless, tích hợp chặt với hệ sinh thái data lake của AWS (S3, Glue Catalog)
Azure Data Factory	Orchestration + ETL/ELT managed trên Azure	Trình thiết kế pipeline trực quan (visual), bộ connector rộng, tích hợp với Azure Synapse
Google Cloud Dataflow	Xử lý batch/stream managed (Apache Beam)	Mô hình thống nhất batch và streaming, tích hợp chặt với BigQuery/Pub/Sub

Các tool này giải quyết extraction và loading một cách đáng tin cậy (tự xử lý pagination, rate limit, schema drift, và retry cho bạn) nhưng cố tình giữ mỏng phần transformation, để “T” lại cho các tool warehouse-native như dbt. Việc điều phối khi nào các ingestion job này chạy, và cách chúng nối vào các job transformation phía sau (downstream), là mối quan tâm của một workflow orchestrator, được trình bày sâu hơn trong ./09-workflow-orchestration.md.

Best Practices

• Mặc định chọn ELT trừ khi có lý do cụ thể không nên làm vậy. Với pipeline đổ vào một cloud warehouse hiện đại, hãy load dữ liệu thô trước và transform bằng SQL/dbt; chỉ dùng transformation trước khi load khi có một ràng buộc thực sự bắt buộc — ví dụ yêu cầu pháp lý phải loại bỏ PII trước khi lưu trữ ở bất cứ đâu, hoặc một đích đến thực sự quá hạn chế để hỗ trợ transformation tại chỗ.
• Mặc định làm cho mọi lần load đều idempotent, sử dụng ngữ nghĩa MERGE/upsert trên một khóa ổn định thay vì INSERT mù quáng. Coi “liệu tôi có thể chạy lại cái này an toàn không?” là một thuộc tính bắt buộc của mọi bước load, không phải một tối ưu hóa để thêm sau.
• Checkpoint các job incremental trên một watermark bền vững, có thể query được (một cột timestamp hoặc offset, được lưu bên ngoài bộ nhớ của chính job), và đảm bảo watermark chỉ tiến lên sau khi một batch được xác nhận đã ghi hoàn tất — tiến watermark quá sớm sẽ tạo ra lỗ hổng nếu một bước sau đó fail.
• Giữ lại dữ liệu thô. Không bao giờ transform theo cách loại bỏ bản extract gốc; hãy đổ nó nguyên vẹn trước tiên (phần “L” trong ELT) để bạn luôn có một nguồn sự thật để replay hoặc diff khi phát hiện một lỗi transformation.
• Phân lớp transformation (staging → intermediate → marts) thay vì viết một câu query lớn làm tất cả mọi việc. Mỗi lớp nên có một nhiệm vụ rõ ràng duy nhất, có thể test độc lập, và được tham chiếu bằng ref()/tương đương để dependency graph luôn tường minh thay vì ngầm định.
• Đừng tự xây integration Reverse ETL tùy chỉnh trước khi kiểm tra xem một tool như Hightouch hay Census đã hỗ trợ đích đó chưa — giá trị của Reverse ETL nằm ở việc biến warehouse thành nguồn sự thật duy nhất, và việc tái phát minh field mapping, lịch đồng bộ, và xử lý rate limit cho từng đích thường không đáng công sức kỹ thuật bỏ ra.
• Test transformation, không chỉ extraction. Một pipeline extract và load hoàn hảo nhưng âm thầm làm hỏng một khóa MERGE hoặc làm rớt row trong một join thì cũng hỏng chẳng khác gì một pipeline không bao giờ chạy được; xem ./16-testing-for-data-pipelines.md để biết dbt test và data quality check phù hợp vào đâu trong bức tranh này.

Tài liệu tham khảo

• dbt Docs — What is dbt?
• dbt Docs — Models
• dbt Docs — Ref and source functions
• Fivetran — What is ELT?
• Airbyte Docs
• Hightouch — What is Reverse ETL?
• Census — Reverse ETL Guide
• Snowflake — MERGE statement reference

Source ──▶ Extract ──▶ Transform (external engine) ──▶ Load ──▶ Destination

Source ──▶ Extract ──▶ Load (raw) ──▶ Destination ──▶ Transform (in-warehouse SQL)

Warehouse (modeled data) ──▶ Reverse ETL tool ──▶ Operational SaaS tool (Salesforce, HubSpot, Zendesk...)

-- models/staging/stg_orders.sql
-- Staging: 1:1 with a raw source table, light renaming/casting only, no joins.
with source as (
    select * from {{ source('ecommerce', 'orders') }}
),
renamed as (
    select
        id as order_id,
        customer_id,
        cast(ordered_at as timestamp) as ordered_at,
        status,
        amount_cents / 100.0 as amount_usd
    from source
)
select * from renamed

-- models/intermediate/int_orders_joined_customers.sql
-- Intermediate: joins/business logic that isn't yet a final reporting shape.
with orders as (
    select * from {{ ref('stg_orders') }}
),
customers as (
    select * from {{ ref('stg_customers') }}
)
select
    orders.order_id,
    orders.ordered_at,
    orders.amount_usd,
    customers.customer_id,
    customers.signup_date,
    customers.country
from orders
left join customers on orders.customer_id = customers.customer_id

-- models/marts/fct_orders.sql
-- Mart: the final, business-facing table analysts and BI tools query directly.
select
    order_id,
    customer_id,
    country,
    date_trunc('month', ordered_at) as order_month,
    amount_usd
from {{ ref('int_orders_joined_customers') }}
where status = 'completed'

-- NOT idempotent: rerunning this after a partial failure
-- duplicates every row that made it through the first time.
insert into orders_raw (order_id, customer_id, amount_usd, ordered_at)
select order_id, customer_id, amount_usd, ordered_at
from staging_orders_batch;

-- Idempotent: PostgreSQL upsert. Reruns overwrite existing rows
-- instead of duplicating them.
insert into orders_raw (order_id, customer_id, amount_usd, ordered_at)
select order_id, customer_id, amount_usd, ordered_at
from staging_orders_batch
on conflict (order_id)
do update set
    customer_id = excluded.customer_id,
    amount_usd = excluded.amount_usd,
    ordered_at = excluded.ordered_at;

-- Idempotent: MERGE (Snowflake/BigQuery/Databricks syntax).
merge into orders_raw as target
using staging_orders_batch as source
on target.order_id = source.order_id
when matched then update set
    customer_id = source.customer_id,
    amount_usd = source.amount_usd,
    ordered_at = source.ordered_at
when not matched then insert (order_id, customer_id, amount_usd, ordered_at)
    values (source.order_id, source.customer_id, source.amount_usd, source.ordered_at);

-- A watermark table tracks the last successfully processed timestamp per source.
create table pipeline_watermarks (
    source_name text primary key,
    last_loaded_at timestamp not null
);

-- Extraction query only pulls rows newer than the last successful watermark.
select *
from source_orders
where updated_at > (
    select last_loaded_at from pipeline_watermarks where source_name = 'orders'
);

-- After a successful load, advance the watermark to the max timestamp just processed.
update pipeline_watermarks
set last_loaded_at = (select max(updated_at) from staging_orders_batch)
where source_name = 'orders';