Menjadwalkan Pipeline dengan Apache Airflow

Ini artikel contoh untuk pembelajaran. Tanggal publikasi dan catatan transaksi bersifat ilustratif.

Scheduler mengatur waktu mulai dan tugas yang harus selesai lebih dahulu. Scheduler tidak menentukan apakah penulisan ulang aman. Contoh Airflow 3.1.6 ini memakai antarmuka publik airflow.sdk untuk menunjukkan urutan dependensi dengan data kecil di memori.

Tetapkan dependensi batch

from datetime import datetime, timedelta, timezone
from airflow.sdk import DAG, task

with DAG(
    dag_id='sample_transaction_check',
    start_date=datetime(2026, 1, 1, tzinfo=timezone.utc),
    schedule='@daily',
    catchup=False,
    default_args={'retries': 2,
                  'retry_delay': timedelta(minutes=2)},
) as dag:
    @task
    def extract():
        return [{'transaction_id': 'T-10', 'amount': '12.50'}]

    @task
    def check(rows):
        if any(not row['transaction_id'] for row in rows):
            raise ValueError('Missing transaction identifier')
        return len(rows)

    @task
    def publish(row_count):
        print(f'Validated sample rows: {row_count}')

    publish(check(extract()))

Nilai keluaran tugas membentuk dependensi: check menunggu extract dan publish menunggu check. Contoh hanya mencetak jumlah baris. Pada pipeline nyata, publikasi mempromosikan batch tervalidasi atau melakukan upsert ke tujuan. Penulisan itu harus tahan percobaan ulang tanpa menambah duplikat.

Pahami interval dan percobaan ulang

Pada Airflow 3.1.6, jadwal cron seperti @daily memakai trigger timetable secara default karena create_cron_data_intervals bernilai False. Jangan menganggap contoh otomatis mewakili batch hari sebelumnya. Timetable berbasis interval data menjadwalkan run setelah intervalnya selesai. Pilih timetable tersebut atau jendela batch tetap yang eksplisit jika sumber membutuhkan input harian terbatas.

start_date adalah batas penjadwalan, bukan perintah agar tugas langsung berjalan pada waktu tersebut. Pilih catatan sumber memakai jendela tetap yang disepakati. Waktu saat ini dapat menghasilkan data berbeda pada percobaan ulang.

retries=2 mengizinkan dua percobaan ulang setelah upaya pertama gagal. Timeout jaringan mungkin pulih sendiri; pelanggaran skema biasanya memerlukan perbaikan sumber. Catat identifier batch dan penyebab agar operator bisa membedakannya. Jangan mengubah kegagalan validasi menjadi batch kosong yang dilaporkan berhasil.

catchup=False mencegah scheduler otomatis membuat semua interval historis ketika DAG diaktifkan. Pengaturan itu tidak memperbaiki data masa lalu yang hilang. Rencanakan backfill eksplisit dengan interval yang diperiksa bila pemulihan membutuhkannya.

Batasi ukuran pesan tugas

List keluaran dikirim melalui mekanisme pesan tugas Airflow dan hanya sesuai untuk demonstrasi ini. Untuk data nyata, kirim lokasi penyimpanan tahan gagal beserta metadata batch. Simpan kredensial pada sistem connection atau secret deployment, bukan di sumber DAG. Uji penjadwalan, import, dan tujuan pada instalasi Airflow Anda sebelum mengaktifkan penulisan.