dlt Python 2026: ELT Pipeline Guide

Q: dlt xử lý schema evolution như thế nào?

dlt có tham số schema_contract với ba chế độ: evolve (thêm cột tự do, mặc định), freeze (chặn thay đổi, raise error), và discard_row / discard_value . Khuyến nghị dùng freeze cho column và data_type ở pipeline production để bắt schema drift ngay khi vendor API đổi.

Q: Deploy dlt lên Airflow như thế nào?

Đơn giản nhất là dùng PythonOperator gọi trực tiếp pipeline.run() . Với Airflow 2.9+, có provider apache-airflow-providers-dlt cung cấp DltRunOperator tự nhận Load Info và emit metric. Với Dagster, dùng dagster-dlt để tạo asset tự động từ dlt source.

Cập nhật: 12 tháng 7, 2026

dlt (data load tool) là thư viện Python open-source cho phép bạn viết ELT pipeline theo phong cách declarative. Bạn chỉ khai báo nguồn dữ liệu, còn dlt sẽ tự lo phần schema inference, incremental loading, state management và ghi vào warehouse như BigQuery, Snowflake, Postgres hoặc DuckDB. Sang 2026, dlt đã trở thành lựa chọn mặc định của nhiều data team muốn thoát khỏi Airbyte connector đắt tiền, hoặc pipeline SQL viết tay dễ vỡ mỗi khi source API thay đổi. Bài này tôi sẽ dẫn bạn đi từ pipeline đầu tiên đến những best practice tôi rút ra sau khi vận hành dlt trên production hơn một năm.

dlt 1.x (phát hành 2024, ổn định 2025-2026) là ELT framework Python declarative với schema inference tự động, hỗ trợ incremental append/replace/merge và hơn 30 destination.
Chỉ cần @dlt.resource và pipeline.run(), dlt sẽ lo phần chuẩn hóa JSON lồng nhau, tạo bảng, mở rộng cột và giữ state giữa các lần chạy.
Merge disposition với primary_key và merge_key giúp xử lý CDC (change data capture) mà không cần viết SQL upsert phức tạp cho từng warehouse.
dlt tích hợp mượt với dbt Core và dbt Cloud thông qua dlt.dbt.package(), biến ELT + transformation thành một orchestrated workflow duy nhất.
Data contract bằng Pydantic model chặn schema drift trước khi nó phá report. Đây là tính năng khiến dlt vượt trội so với script Airbyte connector.
Chi phí vận hành thấp: dlt chạy in-process (Python function), không cần cluster riêng như Airbyte hay Meltano, phù hợp với team nhỏ và pipeline serverless trên Cloud Run hoặc Lambda.

dlt là gì và tại sao data team chuyển sang dùng năm 2026?

dlt (viết tắt của data load tool, luôn viết thường) là một thư viện Python declarative do dltHub phát triển. Thay vì bạn phải tự viết logic chunking, schema DDL, retry, state tracking và upsert SQL cho từng destination, bạn chỉ khai báo một Python generator trả về dict hoặc dataclass. Phần còn lại dlt lo hết. Kể từ khi ra bản 1.0 vào cuối 2024, dlt đã có API ổn định và được adopt bởi các team từ startup đến enterprise như Untitled Data Company, Taktile và Hopper.

Có ba lý do khiến 2026 là năm dlt bùng nổ ở Việt Nam và khu vực APAC. Thứ nhất, xu hướng ELT thay ETL đã hoàn tất. Mọi người đổ dữ liệu thô vào warehouse rồi transform bằng dbt, và dlt là mảnh ghép "EL" gọn nhẹ nhất. Thứ hai, chi phí Airbyte Cloud tăng gấp đôi trong hai năm khiến các team tìm giải pháp thay thế; dlt open-source, chạy in-process nên chi phí gần như bằng không. Thứ ba, khi Iceberg trở thành table format tiêu chuẩn của lakehouse (Databricks mua Tabular năm 2024 xác nhận điều này), việc có một Python library ghi trực tiếp vào Iceberg mà không cần Spark là tính năng critical, và dlt hỗ trợ Iceberg destination native.

Theo kinh nghiệm cá nhân, thứ khiến dlt khác biệt là pipeline test được. Vì mọi thứ là Python function, tôi có thể viết pytest chạy pipeline với DuckDB local rồi assert bảng đầu ra. Đây là điều mà Airbyte connector viết bằng Java hoặc script cron viết tay không cho phép. Cũng là điểm tôi luôn nhấn mạnh khi review kiến trúc pipeline mới cho team.

Cài đặt dlt và khởi tạo project

dlt phân phối qua PyPI với extras cho từng destination. Cài đặt cho DuckDB (local dev) và Postgres (staging):

# Cần Python 3.9 trở lên. Khuyến nghị 3.11 hoặc 3.12 cho performance
pip install "dlt[duckdb,postgres]==1.4.*"

# Kiểm tra version
dlt --version
# dlt 1.4.0

# Khởi tạo project với source template
dlt init github duckdb

Câu lệnh dlt init <source> <destination> tạo ra scaffold với ba file quan trọng: github_pipeline.py (script chạy pipeline), .dlt/config.toml (config không nhạy cảm) và .dlt/secrets.toml (credentials, không commit vào git). Tôi luôn thêm .dlt/secrets.toml vào .gitignore ngay lập tức, vì đây là footgun đầu tiên bạn cần tránh, template mặc định đôi khi commit sample values.

Cấu trúc secrets.toml cho Postgres destination:

# .dlt/secrets.toml
[destination.postgres.credentials]
database = "warehouse"
username = "dlt_loader"
password = "***"
host = "warehouse.internal"
port = 5432

[sources.github]
access_token = "ghp_***"

Xây dựng pipeline đầu tiên: REST API đến DuckDB

Ví dụ dưới đây pull issues từ GitHub REST API rồi ghi vào DuckDB. Đây là pipeline mẫu tôi hay dùng khi onboard junior engineer, vì nó cover đủ 4 concept: resource, source, pipeline và disposition.

import dlt
from dlt.sources.helpers import requests

@dlt.resource(
    write_disposition="merge",
    primary_key="id",
    name="issues",
)
def github_issues(repo: str, cursor_updated=dlt.sources.incremental("updated_at")):
    """Lấy issue theo trang, chỉ pull bản ghi mới hơn cursor."""
    url = f"https://api.github.com/repos/{repo}/issues"
    params = {
        "state": "all",
        "per_page": 100,
        "since": cursor_updated.last_value,
        "sort": "updated",
        "direction": "asc",
    }
    while url:
        r = requests.get(url, params=params)
        r.raise_for_status()
        yield r.json()
        url = r.links.get("next", {}).get("url")
        params = None  # link header đã chứa params

if __name__ == "__main__":
    pipeline = dlt.pipeline(
        pipeline_name="github_issues_prod",
        destination="duckdb",
        dataset_name="raw_github",
    )
    info = pipeline.run(github_issues("pola-rs/polars"))
    print(info)

Chạy file, dlt sẽ: tạo file github_issues_prod.duckdb, tự infer schema từ JSON, chuẩn hóa JSON lồng (ví dụ user.login thành cột user__login), tạo bảng con issues__labels cho array và ghi state cursor vào bảng _dlt_pipeline_state. Lần chạy tiếp theo chỉ pull issue có updated_at lớn hơn cursor cuối. Logic incremental bạn không phải viết một dòng.

Nếu bạn từng nhìn metric pipeline của mình, chắc bạn biết cảm giác pull lại 2 triệu row mỗi đêm chỉ vì không có cursor. Thật lòng mà nói, dlt giải phóng bạn khỏi điều đó ngay từ resource đầu tiên.

Incremental loading: append, replace và merge

dlt có ba write_disposition ứng với ba pattern chính data engineer gặp hàng ngày. Hiểu đúng ba disposition này quyết định chất lượng pipeline của bạn:

append: cộng dồn bản ghi mới. Phù hợp event log, click stream, transaction ledger, nơi mỗi bản ghi là bất biến.
replace: xóa bảng và ghi lại. Dùng cho reference table nhỏ như "danh sách country". Đơn giản, an toàn, nhưng đắt với bảng lớn.
merge: upsert theo primary_key, xóa mềm bản ghi biến mất (nếu có merge_key). Đây là workhorse cho CDC.

Ví dụ merge với hai tầng key, hữu ích khi bạn pull user profile mà API trả cả bản ghi bị soft-deleted:

@dlt.resource(
    write_disposition="merge",
    primary_key=["tenant_id", "user_id"],
    merge_key=["tenant_id"],  # xóa mềm trong scope tenant
)
def users(tenant_id: str, api_client):
    yield from api_client.list_users(tenant_id)

Schema evolution và cách dlt xử lý cột mới an toàn

Backfill vì API thêm một field mới là nỗi ám ảnh của mọi data engineer. dlt xử lý schema evolution qua ba chế độ schema_contract: evolve (mặc định, thêm cột tự do), freeze (drop bản ghi có cột mới), và discard_row hoặc discard_value.

@dlt.resource(
    name="orders",
    schema_contract={
        "tables": "evolve",       # cho phep bang moi
        "columns": "freeze",      # chan cot moi - alert khi API doi
        "data_type": "freeze",    # chan type change (int -> string)
    },
)
def orders():
    yield from fetch_orders()

Tôi khuyến nghị freeze cho column và data_type trên mọi bảng có downstream dbt model. Lý do rất practical: khi vendor API bất ngờ đổi amount từ integer sang string, pipeline evolve sẽ âm thầm tạo cột amount__v_text và tất cả dbt model xuống dòng sẽ sai kết quả. Freeze sẽ raise DataValidationError, bạn được alert ngay lập tức thay vì phát hiện sau một tuần.

Kết hợp với Pandera cho validation DataFrame trong pipeline, bạn có hai lớp bảo vệ: dlt chặn schema drift ở lớp EL, Pandera assert business rule ở lớp transform. Chiến lược này tôi đã áp dụng cho mọi pipeline financial và chưa bao giờ để lọt một row corrupt xuống warehouse. Không phải khoe, chỉ là có nhiều đêm ngủ ngon hơn.

Data contract với Pydantic model

Từ dlt 1.2, bạn có thể khai báo resource với columns là một Pydantic v2 model. dlt sẽ validate mọi row theo model đó và fail sớm nếu vi phạm. Về bản chất, đây là "type-checked EL".

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional

class Order(BaseModel):
    order_id: str = Field(..., min_length=1)
    tenant_id: str
    amount_cents: int = Field(..., ge=0)
    currency: str = Field(..., pattern="^[A-Z]{3}$")
    placed_at: datetime
    canceled_at: Optional[datetime] = None

@dlt.resource(
    name="orders",
    primary_key="order_id",
    write_disposition="merge",
    columns=Order,
)
def orders(api):
    for raw in api.iter_orders():
        yield raw  # dlt validate moi row bang Order.model_validate()

Đây là contract test tôi luôn thêm vào PR review. Nếu ai đó pull field mới mà quên update model, CI test fail, không có sự cố nào âm thầm lên production nữa. Nó cũng giúp onboard người mới: model chính là tài liệu source of truth cho shape của data raw.

Tích hợp dbt: dlt cho EL, dbt cho T

dlt không thay thế dbt, nó bổ sung. dlt phụ trách phần "extract-load" đưa raw data vào warehouse, còn dbt phụ trách phần "transform" biến raw thành model business-ready. Từ dlt 0.5, có helper dlt.dbt.package() orchestrate cả hai trong cùng một Python script:

import dlt
from dlt.helpers.dbt import package

pipeline = dlt.pipeline(
    pipeline_name="orders_elt",
    destination="bigquery",
    dataset_name="raw",
)

# 1. Extract-load bang dlt
pipeline.run(orders_source())

# 2. Chay dbt project ngay sau do, dung cung credentials
dbt = package(
    pipeline,
    package_location="./dbt/orders_dbt",
    package_profile_name="orders",
)
models = dbt.run_all()  # dbt run + test tat ca model
for m in models:
    print(m.model_name, m.status, m.message)

Điểm ăn tiền là dlt share credentials với dbt profile tự động, bạn không phải maintain hai bộ secret cho cùng một warehouse. Trong pipeline production, tôi cho dbt.run_all() chạy sau pipeline.run() ở cùng một Airflow task; nếu step nào fail, cả DAG fail và alert. Cách này giải quyết vấn đề dài kỳ của EL và T bị "lệch trạng thái", vốn rất phổ biến khi Airbyte và dbt Cloud chạy độc lập.

Bạn cũng có thể kết hợp thêm DuckDB làm engine transform local để test dbt model không cần đụng warehouse. Đây là pattern development loop nhanh nhất tôi từng dùng.

Destinations phổ biến: BigQuery, Snowflake, Postgres, Iceberg

dlt 1.4 hỗ trợ hơn 30 destination. Bốn cái tôi gặp thường xuyên nhất trong prod:

Destination	Use case	Merge support	Note
DuckDB	Dev, test, ad-hoc analytics	Full	Zero setup, pipeline test-run trong pytest
BigQuery	Warehouse chính cho analytics	Full (via MERGE)	Streaming insert cho low-latency; batch cho cost
Snowflake	Enterprise warehouse	Full	Hỗ trợ external table trên S3
Postgres	Operational store, staging	Full	Dễ debug nhất, phù hợp small data
Iceberg (via PyIceberg)	Lakehouse open	Append + merge	Ghi trực tiếp không cần Spark
Filesystem (Parquet)	Data lake raw zone	Append only	Kết hợp với Athena/Trino

Với team đang chuyển sang lakehouse, kết hợp dlt + Iceberg + Ibis cho DataFrame API đa backend tạo thành stack Python-first hoàn chỉnh, không cần Spark. Đây là combo tôi khuyến nghị cho các startup không đủ ngân sách Databricks nhưng muốn giữ open format.

So sánh dlt với Airbyte và Meltano

Câu hỏi tôi bị hỏi mỗi tuần: "Nên chọn dlt, Airbyte hay Meltano?" Không có câu trả lời duy nhất, nhưng có tiêu chí lựa chọn rõ ràng:

Tiêu chí	dlt	Airbyte	Meltano
Ngôn ngữ định nghĩa	Python function	YAML/UI + Java/Python connector	YAML meltano.yml
Deployment model	In-process (bất kỳ Python runtime)	K8s cluster hoặc Cloud	Container + Airflow/Dagster
Số connector sẵn	~30 verified + REST/SQL generic	350+	500+ Singer taps
Test local	pytest thuần	Docker connector test	meltano test
Chi phí	Free, self-host	Free OSS / Cloud từ ~$2.5/GB	Free, self-host
Code contract	Pydantic native	JSON schema	Singer schema JSON
Learning curve	Thấp cho Python dev	Trung bình	Cao (Singer spec)

Chọn dlt khi team bạn là Python developer và cần custom logic (auth phức tạp, pagination lạ). Chọn Airbyte khi cần integrate 200+ SaaS mà không muốn viết code, trade-off là chi phí runtime cao hơn. Meltano phù hợp team đã đầu tư vào Singer ecosystem. Với 90% pipeline internal tôi từng thấy, dlt là lựa chọn đúng, vì source thường là API nội bộ hoặc Postgres CDC chứ không phải Salesforce/HubSpot.

Best practice production: testing, monitoring, deploy

dlt cho phép bạn viết pipeline test được, và test là non-negotiable cho production. Đây là setup pytest tôi dùng cho mọi dlt pipeline:

import dlt
import pytest

@pytest.fixture
def test_pipeline(tmp_path):
    return dlt.pipeline(
        pipeline_name="test",
        destination="duckdb",
        dataset_name="raw",
        pipelines_dir=str(tmp_path),  # tach state khoi env dev
    )

def test_issues_merge_dedup(test_pipeline, mock_api):
    """Chay resource hai lan voi overlap, assert khong co duplicate."""
    test_pipeline.run(github_issues("acme/repo"))
    test_pipeline.run(github_issues("acme/repo"))
    with test_pipeline.sql_client() as c:
        row = c.execute_sql(
            "SELECT COUNT(*) - COUNT(DISTINCT id) FROM raw.issues"
        )[0][0]
        assert row == 0

Ba nguyên tắc tôi tuân thủ: (1) luôn pipelines_dir=tmp_path để test cô lập; (2) mock HTTP với responses hoặc vcrpy, không call live API trong CI; (3) assert count DISTINCT thay vì count thô để bắt bug merge disposition.

Về deploy, dlt chạy trên bất cứ đâu Python chạy được: Airflow PythonOperator, Dagster asset, Prefect flow, Cloud Run job, Lambda. Với schedule đơn giản, tôi hay dùng GitHub repo dlt-hub cùng GitHub Actions cron. Pipeline nhỏ chạy ổn định vài tháng liền không sự cố. Ngoài ra, khuôn khổ OpenLineage được chuẩn hóa qua OpenLineage spec, dlt 1.3 tuân thủ chuẩn này khá đầy đủ.

Quan sát pipeline: logs, load info và trace

Mỗi lần pipeline.run() trả về LoadInfo chứa số row load được, timing từng bước, và ID của job trên destination. Trong production tôi log object này vào một bảng _pipeline_runs để dashboard Grafana theo dõi:

info = pipeline.run(source)
metrics = {
    "pipeline": info.pipeline.pipeline_name,
    "started_at": info.started_at.isoformat(),
    "finished_at": info.finished_at.isoformat(),
    "load_id": info.loads_ids[0],
    "rows_loaded": sum(p.metrics[0]["job_metrics"][j]["items_count"]
                       for p in info.load_packages
                       for j in p.metrics[0]["job_metrics"]),
}
publish_to_grafana(metrics)

Alert khi rows_loaded == 0 nhiều lần liên tiếp. Đây là cách tôi bắt được một incident vendor API im lặng downtime tuần trước. Đừng tin pipeline nào chạy "xanh" mà không có bằng chứng row count hợp lý, thật đấy.

Những pitfall cần tránh khi mới dùng dlt

Sau khi giúp bốn team migrate từ Airbyte và cron script sang dlt, đây là ba lỗi lặp lại nhiều nhất:

Quên đặt primary_key khi merge: dlt sẽ silent chuyển sang append và bạn có duplicate. Luôn assert primary key trong PR review.
Đặt full_refresh=True cho pipeline cursor: mất state, pull lại toàn bộ. Chỉ dùng flag này lúc dev.
Bỏ qua staging destination cho warehouse lớn: BigQuery/Snowflake load nhanh hơn nhiều khi qua GCS/S3 staging. Thêm staging="filesystem" cho bảng >100M row.

Nói thật, tôi đã dính cả ba lỗi này ít nhất một lần trong năm đầu dùng dlt. Không có gì đáng xấu hổ, chỉ cần bạn học được là ổn (và giờ bạn đã học rồi, đường ngắn hơn tôi đi).

Câu hỏi thường gặp

dlt có thay thế được dbt không?

Không. dlt lo phần "EL" (extract-load) đưa raw data vào warehouse, còn dbt lo phần "T" (transform) biến raw thành model business. Hai công cụ này bổ sung nhau; dlt còn có helper dlt.dbt.package() để orchestrate cả EL và T trong một script Python duy nhất.

dlt khác Airbyte ở điểm nào?

dlt là Python library chạy in-process, không cần cluster K8s riêng như Airbyte. Bạn định nghĩa pipeline bằng Python function nên test dễ và tùy biến tự do. Airbyte có sẵn nhiều connector SaaS hơn, phù hợp khi không muốn viết code. dlt phù hợp team Python với source API custom.

dlt xử lý schema evolution như thế nào?

dlt có tham số schema_contract với ba chế độ: evolve (thêm cột tự do, mặc định), freeze (chặn thay đổi, raise error), và discard_row/discard_value. Khuyến nghị dùng freeze cho column và data_type ở pipeline production để bắt schema drift ngay khi vendor API đổi.

dlt có hỗ trợ Iceberg và lakehouse không?

Có. Từ 2025, dlt hỗ trợ Iceberg destination thông qua PyIceberg, cho phép ghi trực tiếp vào Iceberg table trên S3/GCS mà không cần cluster Spark. Điều này rất phù hợp cho team đang xây lakehouse Python-first, kết hợp với Ibis hoặc Trino cho query layer.

dlt có production-ready cho enterprise workload không?

Có. dlt 1.x đã ổn định từ cuối 2024 và được adopt bởi Untitled, Taktile, Hopper cùng nhiều enterprise. Nó có test suite đầy đủ, OpenLineage integration, staging table cho atomic swap, và cost of ownership thấp vì chạy in-process. Tôi đã vận hành dlt trên pipeline financial hơn một năm mà không gặp sự cố nghiêm trọng.

Deploy dlt lên Airflow như thế nào?

Đơn giản nhất là dùng PythonOperator gọi trực tiếp pipeline.run(). Với Airflow 2.9+, có provider apache-airflow-providers-dlt cung cấp DltRunOperator tự nhận Load Info và emit metric. Với Dagster, dùng dagster-dlt để tạo asset tự động từ dlt source.