Berkas .dlt/secrets.toml otomatis ditambahkan ke .gitignore. Jangan pernah meng-commit-nya.

Konsep Inti: Resource, Source, Pipeline

dlt memiliki tiga blok bangunan yang perlu dipahami sebelum menulis kode.

• Resource (@dlt.resource): fungsi yang menghasilkan data — biasanya baris atau halaman (page) record. Setiap resource dipetakan ke satu tabel di destinasi.
• Source (@dlt.source): fungsi yang mengelompokkan beberapa resource terkait (misalnya semua endpoint dari satu API). Ia mengembalikan resource yang dikelolanya.
• Pipeline (dlt.pipeline(...)): objek yang menghubungkan source ke destinasi, menjalankan pemuatan, dan melacak state (skema, cursor inkremental, riwayat run).

Pipeline Pertama: REST API Berpaginasi ke DuckDB

Mari memuat issue dari REST API berpaginasi ke DuckDB. Kita pakai REST API GitHub sebagai contoh berjalan karena paginasinya mewakili kebanyakan API. Perhatikan bagaimana resource, pengelompokan ala source, dan pipeline menyatu: dlt membuat berkas DuckDB, menyimpulkan skema dari data, lalu menulis barisnya tanpa DDL atau pembuatan tabel manual.

Beberapa hal yang perlu dicatat:

Untuk DuckDB Anda juga bisa membuka berkas .duckdb yang dihasilkan langsung di klien DuckDB apa pun dan menjalankan SQL seperti SELECT id, title, state FROM githubdata.issues ORDER BY createdat DESC LIMIT 10;.

SELECT l.name, i.title
FROM issueslabels l

JOIN issues i ON l.dltparentid = i.dltid;

dlt menambahkan kolom pembukuan berprefiks dlt (dltid, dltloadid, dltparentid) agar relasi dan batch pemuatan ini dapat ditelusuri.

Evolusi

Saat sumber menambah field baru, dlt mendeteksinya pada run berikutnya dan otomatis menambahkan kolom ke tabel destinasi — tanpa mematahkan pemuatan. Field yang dihapus cukup menghasilkan NULL untuk baris baru. Evolusi skema inilah yang membuat dlt tahan terhadap perubahan API upstream yang jika tidak demikian akan mematahkan loader buatan tangan.

Anda bisa memeriksa skema yang disimpulkan kapan saja.

print(pipeline.defaultschema.toprettyyaml())

# replace: hapus dan muat ulang tabel tiap run (full refresh)
@dlt.resource(writedisposition="replace")

def dimcountry(): ...

append: tambah baris baru, tidak menyentuh yang lama (cocok untuk event/log yang imutabel)
@dlt.resource(writedisposition="append")

def pageviews(): ...

merge: upsert — perbarui baris yang ada berdasarkan key, sisipkan yang baru
@dlt.resource(writedisposition="merge", primarykey="id")
def orders(): ...

• replace paling sederhana dan tepat untuk tabel dimensi kecil yang masih layak dibangun ulang.
• append untuk fakta yang tidak pernah berubah, seperti log event. Hati-hati duplikat saat run ulang.
• merge adalah upsert. Anda deklarasikan primarykey (atau mergekey); dlt menghapus baris yang cocok dan menyisipkan batch masuk, sehingga run ulang bersifat idempoten.

Gunakan mergekey alih-alih primarykey ketika natural key terdiri dari beberapa kolom atau ketika Anda ingin mengatur predikat penghapusan secara terpisah dari identitas baris.

Pemuatan Inkremental dengan dlt.sources.incremental

Muat ulang penuh tidak skalabel. Pemuatan inkremental hanya mengambil record yang berubah sejak run terakhir, menggunakan cursor field (kolom yang naik secara monoton seperti updatedat atau id).

import dlt
from dlt.sources.helpers import requests

@dlt.resource(tablename="issues", writedisposition="merge", primarykey="id")
def githubissues(

    repo: str = "dlt-hub/dlt",
    updatedat=dlt.sources.incremental(
        "updatedat",

        initialvalue="2024-01-01T00:00:00Z",
    ),
):
    url = f"https://api.github.com/repos/{repo}/issues"
    params = {
        "state": "all",
        "perpage": 100,

        "page": 1,
        "since": updatedat.lastvalue,   # minta hanya data baru ke API

        "sort": "updated",
        "direction": "asc",
    }
    while True:
        response = requests.get(url, params=params)
        response.raiseforstatus()

        page = response.json()
        if not page:
            break
        yield page
        params["page"] += 1

• Pada run pertama, updatedat.lastvalue sama dengan initialvalue. dlt memuat semuanya sejak titik itu.
• dlt menyimpan nilai updatedat terbesar yang dilihatnya di state pipeline.
• Pada run berikutnya, lastvalue adalah nilai maksimum tersimpan itu, sehingga Anda hanya mengambil record yang lebih baru.
• Dikombinasikan dengan writedisposition="merge" dan primarykey="id", record yang diperbarui di upstream akan di-upsert, bukan diduplikasi.

dlt juga mendeduplikasi baris pada nilai batas (record dengan lastvalue persis) agar tepi data tidak termuat ganda. State dipertahankan di destinasi, sehingga bertahan lintas proses run yang terpisah — tanpa perlu penyimpanan state eksternal.

Menulis loop paginasi secara manual itu berulang dan melelahkan. dlt menyediakan helper deklaratif restapisource yang mendeskripsikan API sebagai konfigurasi. Ia menangani paginasi, autentikasi, dan penyambungan inkremental untuk Anda.

import dlt
from dlt.sources.restapi import restapisource

source = restapisource({
    "client": {
        "baseurl": "https://api.github.com/",

        "auth": {
            "type": "bearer",
            "token": dlt.secrets["githubtoken"],
        },
        "paginator": {
            "type": "headerlink",          # GitHub memakai Link header

        },
    },
    "resourcedefaults": {
        "primarykey": "id",

        "writedisposition": "merge",
    },
    "resources": [
        {
            "name": "issues",
            "endpoint": {
                "path": "repos/dlt-hub/dlt/issues",
                "params": {
                    "state": "all",
                    "since": {
                        "type": "incremental",
                        "cursorpath": "updatedat",
                        "initialvalue": "2024-01-01T00:00:00Z",

                    },
                },
            },
        },
        {
            "name": "comments",
            "endpoint": {"path": "repos/dlt-hub/dlt/issues/comments"},
        },
    ],
})

pipeline = dlt.pipeline(
    pipelinename="githubrest",

    destination="duckdb",
    datasetname="githubdata",

)
print(pipeline.run(source))

[sources.github]
githubtoken = "ghppersonalaccesstokenanda"

[destination.bigquery]
location = "EU"

[destination.bigquery.credentials]
projectid = "my-gcp-project"

privatekey = "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n"
clientemail = "loader@my-gcp-project.iam.gserviceaccount.com"

[runtime]
loglevel = "INFO"

[load]
workers = 4

Di produksi, utamakan environment variable. dlt memetakan hierarki TOML ke nama huruf besar yang dipisahkan garis bawah ganda.

export SOURCESGITHUBGITHUBTOKEN="ghp..."
export DESTINATIONBIGQUERYCREDENTIALS_PROJECTID="my-gcp-project"

Rujuk sebuah secret di kode dengan dlt.secrets["..."] atau, yang lebih idiomatik, dengan memberi fungsi Anda argumen bertipe dengan nilai default dlt.secrets.value — dlt menyuntikkan nilai yang sudah diselesaikan secara otomatis saat pemanggilan.

@dlt.source
def githubsource(githubtoken: str = dlt.secrets.value):
    ...

Data Contract dan Kontrol Skema

Secara default dlt mengevolusi skema dengan bebas, yang nyaman saat pengembangan tetapi berisiko di produksi. Pengaturan schemacontract memungkinkan Anda membatasi perilaku itu per tabel, kolom, dan tipe data.

@dlt.resource(
    schemacontract={
        "tables": "evolve",   # izinkan tabel baru
        "columns": "freeze",  # tolak kolom baru -> error pada field tak terduga
        "datatype": "freeze" # tolak perubahan tipe

    }
)
def orders():
    ...

Mode yang tersedia adalah evolve (izinkan, default), freeze (lempar error), discardrow (buang baris bermasalah), dan discardvalue (buang nilai bermasalah tetapi pertahankan baris). Pola produksi yang umum adalah evolve untuk kolom tetapi freeze untuk tipe data, sehingga perubahan aditif lolos sementara pergeseran tipe terdeteksi sejak dini.

@dlt.resource
def issueids():
    yield from [{"id": 1}, {"id": 2}, {"id": 3}]

@dlt.transformer(datafrom=issueids)
def issuedetail(issue):

    resp = requests.get(f"https://api.example.com/issues/{issue['id']}")
    yield resp.json()

Salurkan resource ke transformer
pipeline.run(issueids | issuedetail)

@dlt.resource(parallelized=True)
def heavyendpoint():
    ...

Mulailah dengan nilai default dan naikkan jumlah worker hanya ketika sebuah tahap terukur menjadi bottleneck.

Menjalankan dan Men-deploy

Pipeline dlt hanyalah skrip Python, sehingga deployment-nya adalah apa pun yang menjalankan Python secara terjadwal.

Inilah persis sambungan tempat dlt menyerahkan ke dbt: dlt mendaratkan raw, dbt membangun mart di atasnya.

Praktik Terbaik