first commit
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
"""sitomat — DAG-3 (gold build) — TAHAP A: CONNECTIVITY SMOKE DAG (Airflow 3.x).
|
||||
|
||||
HONESTY (baca dulu):
|
||||
DAG ini HANYA memvalidasi konektivitas sitomat <-> Airflow (validate DAG,
|
||||
trigger run, poll status). Ia TIDAK membangun gold dan **TIDAK** memanggil
|
||||
callback `register_gold` — jadi **tidak ada artefak gold yang diklaim**.
|
||||
|
||||
Tahap B akan menggantinya dengan transform nyata: baca silver (via Trino),
|
||||
tulis gold, lalu callback ke sitomat. Lihat README.md di folder yang sama.
|
||||
|
||||
Airflow 3.x: dataclass import publik adalah `airflow.sdk` (bukan
|
||||
`airflow.decorators`/`airflow.models`), dan `schedule_interval` sudah dihapus
|
||||
-> gunakan `schedule=None` untuk DAG yang hanya dipicu lewat REST.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import pendulum
|
||||
from airflow.sdk import dag, get_current_context, task
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Kunci conf yang WAJIB dikirim sitomat saat trigger (lihat F16 `flattenTriggerConf`).
|
||||
REQUIRED_CONF_KEYS = (
|
||||
"correlationId",
|
||||
"dataProductBuildId",
|
||||
"dataProductId",
|
||||
"useCaseId",
|
||||
)
|
||||
|
||||
|
||||
@dag(
|
||||
dag_id="sitomat_gold_build", # samakan dengan env `GOLD_DAG_ID` di sitomat
|
||||
description="sitomat F16 DAG-3 — connectivity smoke (tidak membangun gold)",
|
||||
schedule=None, # Airflow 3: hanya dipicu lewat REST (POST /api/v2/.../dagRuns)
|
||||
start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
|
||||
catchup=False,
|
||||
tags=["sitomat", "f16", "connectivity"],
|
||||
)
|
||||
def sitomat_gold_build():
|
||||
@task
|
||||
def parse_conf() -> dict:
|
||||
"""Baca `dag_run.conf` dari sitomat dan validasi kunci wajib."""
|
||||
ctx = get_current_context()
|
||||
conf = dict((ctx["dag_run"].conf or {})) if ctx else {}
|
||||
|
||||
missing = [key for key in REQUIRED_CONF_KEYS if not conf.get(key)]
|
||||
if missing:
|
||||
# Gagal jujur (BUKAN sukses palsu): conf tidak lengkap.
|
||||
raise ValueError(f"conf sitomat tidak lengkap, hilang: {missing}")
|
||||
|
||||
silver_refs = [ref for ref in str(conf.get("silverRefs", "")).split(",") if ref]
|
||||
log.info(
|
||||
"sitomat conf OK: correlationId=%s build=%s product=%s useCase=%s "
|
||||
"sources=%s silverRefs=%d",
|
||||
conf.get("correlationId"),
|
||||
conf.get("dataProductBuildId"),
|
||||
conf.get("dataProductId"),
|
||||
conf.get("useCaseId"),
|
||||
conf.get("sourceCount"),
|
||||
len(silver_refs),
|
||||
)
|
||||
if not silver_refs:
|
||||
log.warning(
|
||||
"silverRefs kosong — Tahap B (build nyata) belum bisa membaca sumber."
|
||||
)
|
||||
return {
|
||||
"correlationId": conf.get("correlationId"),
|
||||
"dataProductBuildId": conf.get("dataProductBuildId"),
|
||||
"silverRefs": silver_refs,
|
||||
}
|
||||
|
||||
@task
|
||||
def connectivity_report(payload: dict) -> None:
|
||||
"""TAHAP A: sengaja hanya melaporkan. TIDAK menulis gold.
|
||||
|
||||
Tahap B (ganti task ini):
|
||||
1. baca `payload["silverRefs"]` via Trino (koneksi `trino_default`),
|
||||
2. rakit & tulis tabel gold (Iceberg/Nessie),
|
||||
3. POST ke sitomat:
|
||||
{SITOMAT_BASE_URL}/data-products/builds/{dataProductBuildId}/gold
|
||||
header: `X-API-Key: <kunci scope pipeline>`
|
||||
body: {"goldRef": ..., "checksum": ..., "rowCount": ...}
|
||||
Simpan URL & kunci sebagai Airflow Variable/Connection (JANGAN hardcode).
|
||||
"""
|
||||
log.info(
|
||||
"sitomat DAG-3 connectivity OK untuk build=%s (Tahap A: tanpa gold)",
|
||||
payload.get("dataProductBuildId"),
|
||||
)
|
||||
|
||||
connectivity_report(parse_conf())
|
||||
|
||||
|
||||
sitomat_gold_build()
|
||||
@@ -0,0 +1,344 @@
|
||||
"""
|
||||
DAG-3 PRODUKSI — bangun GOLD untuk sebuah Data Product, lalu lapor ke sitomat.
|
||||
|
||||
Fitur F16/F15 (ADR-0004..0008). sitomat TIDAK memindahkan byte: DAG ini yang
|
||||
membangun artefak di data lake (Iceberg lewat Trino) dan memanggil callback.
|
||||
|
||||
Kontrak
|
||||
-------
|
||||
Trigger conf (dikirim F15 `createGoldBuildTriggerPort`; SEMUA nilai STRING):
|
||||
correlationId, dataProductBuildId, dataProductId, useCaseId,
|
||||
datasetVersionIds (dipisah koma), silverRefs (dipisah koma; posisinya
|
||||
sejajar dengan datasetVersionIds, slot kosong = sumber tanpa ref),
|
||||
sourceCount
|
||||
|
||||
Callback WAJIB (tanpa ini build sitomat tetap 'building'):
|
||||
POST {SITOMAT_BASE_URL}/data-products/builds/{dataProductBuildId}/gold
|
||||
Header: X-API-Key: <kunci F09-A scope `pipeline`>
|
||||
Body: {"goldRef": "...", "checksum": "sha256:...", "rowCount": N}
|
||||
|
||||
Konfigurasi — Airflow Variable dulu, lalu env, lalu default:
|
||||
sitomat_base_url SITOMAT_BASE_URL http://172.27.228.103:13000
|
||||
sitomat_api_key SITOMAT_API_KEY (WAJIB) kunci scope `pipeline`
|
||||
trino_url TRINO_URL https://sicabe-trino:8443
|
||||
trino_user TRINO_USER svc_seed
|
||||
trino_password TRINO_PASSWORD (WAJIB)
|
||||
gold_target GOLD_TARGET iceberg.gold.dp_<dataProductId>
|
||||
gold_sql GOLD_SQL opsional; bila kosong dirakit dari silverRefs
|
||||
gold_tls_verify GOLD_TLS_VERIFY false (sertifikat internal self-signed)
|
||||
checksum_max_rows CHECKSUM_MAX_ROWS 100000 (0 = tanpa batas)
|
||||
|
||||
**Presedensi:** `silverRefs` dari conf adalah jalur NYATA dan menang; `gold_sql`
|
||||
= **fallback** untuk dataset yang belum menyimpan `artifact_ref`.
|
||||
|
||||
**Kegagalan DILAPORKAN, bukan menggantung:** build NYATA yang sumbernya belum
|
||||
punya `artifact_ref` (atau tanpa sumber sama sekali) **GAGAL dengan sengaja** —
|
||||
DAG menolak membangun produk parsial dan task `report_failure` memanggil
|
||||
`POST …/fail`, sehingga build/produk/use case menjadi `failed`. Hanya run
|
||||
DIAGNOSTIK (`bun run check:airflow`) yang berhenti diam tanpa callback.
|
||||
|
||||
Cakupan OPA: stack ini memakai allowlist per-tabel (`default allow := false`).
|
||||
Menulis ke tabel gold BARU mensyaratkan policy OPA memberi grup `data-engineer`
|
||||
akses pada schema gold/silver (lihat README). Bila belum, set `GOLD_TARGET` ke
|
||||
tabel yang sudah diizinkan (mis. `iceberg.gold.orders_gold`).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
|
||||
import requests
|
||||
import urllib3
|
||||
from airflow.sdk import dag, get_current_context, task
|
||||
|
||||
# Sertifikat internal self-signed: jangan banjiri log dengan peringatan.
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
|
||||
DEFAULT_BASE_URL = "http://172.27.228.103:13000"
|
||||
DEFAULT_TRINO_URL = "https://sicabe-trino:8443"
|
||||
DEFAULT_TRINO_USER = "svc_seed"
|
||||
DEFAULT_CHECKSUM_MAX_ROWS = 100000
|
||||
|
||||
# Nama identifier SQL yang aman; apa pun di luar ini DITOLAK (anti-injeksi).
|
||||
IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
||||
|
||||
|
||||
# ── Pengaturan (Airflow Variable → env → default) ────────────────────────────
|
||||
|
||||
|
||||
def _variable_get(key: str) -> str | None:
|
||||
"""Baca Airflow Variable `key`; `None` bila tidak ada / tidak tersedia."""
|
||||
Variable = None
|
||||
try:
|
||||
from airflow.sdk import Variable # Airflow 3 SDK
|
||||
except Exception: # pragma: no cover - tergantung versi Airflow
|
||||
try:
|
||||
from airflow.models import Variable # fallback klasik
|
||||
except Exception:
|
||||
return None
|
||||
try:
|
||||
return Variable.get(key)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def setting(name: str, default: str | None = None) -> str | None:
|
||||
"""Airflow Variable `name` (huruf kecil) → env `NAME` (huruf besar) → default."""
|
||||
value = os.environ.get(name.upper())
|
||||
if value:
|
||||
return value
|
||||
variable = _variable_get(name.lower())
|
||||
return variable if variable else default
|
||||
|
||||
|
||||
# ── Util murni ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def qualified(identifier: str) -> str:
|
||||
"""Validasi `tabel` / `schema.tabel` / `catalog.schema.tabel`. Raise bila bukan."""
|
||||
raw = identifier.strip().strip('"').strip("`")
|
||||
parts = [part.strip().strip('"').strip("`") for part in raw.split(".")]
|
||||
if not 1 <= len(parts) <= 3 or not all(IDENTIFIER.match(part) for part in parts):
|
||||
raise ValueError(f"nama tabel tidak valid: {identifier!r}")
|
||||
return ".".join(parts)
|
||||
|
||||
|
||||
def safe_suffix(value: str) -> str:
|
||||
"""Jadikan nilai bebas menjadi sufiks identifier yang aman."""
|
||||
cleaned = re.sub(r"[^A-Za-z0-9_]", "_", value or "x")
|
||||
if not cleaned or cleaned[0].isdigit():
|
||||
cleaned = f"t_{cleaned}"
|
||||
return cleaned
|
||||
|
||||
|
||||
def _as_int(value) -> int:
|
||||
"""Integer non-negatif dari conf (semua nilai conf adalah STRING)."""
|
||||
try:
|
||||
return max(0, int(str(value).strip()))
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
|
||||
|
||||
def target_and_sql(conf: dict, sources: list[str]) -> tuple[str, str]:
|
||||
"""(target_fqn, sql_ctas) dari conf + pengaturan.
|
||||
|
||||
`GOLD_SQL` (Variable `gold_sql`) adalah escape hatch yang disengaja: dipakai
|
||||
bila `silverRefs` kosong (mis. versi silver lama belum menyimpan `artifact_ref`)
|
||||
ATAU untuk mengganti transform. Tanpa keduanya, tidak ada yang bisa dibangun.
|
||||
"""
|
||||
configured = setting("gold_target")
|
||||
target = qualified(
|
||||
configured
|
||||
or f"iceberg.gold.dp_{safe_suffix(str(conf.get('dataProductId', 'x')))}"
|
||||
)
|
||||
explicit = setting("gold_sql")
|
||||
# PRESEDENSI: sumber dari conf (`silverRefs`) adalah jalur NYATA dan menang;
|
||||
# `gold_sql` hanya FALLBACK operator untuk dataset yang belum punya
|
||||
# `artifact_ref`. (Kebalikannya membuat override global menelan lineage.)
|
||||
if sources:
|
||||
select = " UNION ALL ".join(f"SELECT * FROM {source}" for source in sources)
|
||||
elif explicit:
|
||||
select = explicit.strip().rstrip(";")
|
||||
else:
|
||||
raise ValueError("tidak ada silverRefs dan tidak ada gold_sql — tidak ada yang dibangun")
|
||||
return target, f"CREATE OR REPLACE TABLE {target} AS {select}"
|
||||
|
||||
|
||||
# ── Klien Trino REST minimal (/v1/statement + paging nextUri) ─────────────────
|
||||
|
||||
|
||||
class Trino:
|
||||
"""Klien Trino HTTPS berbatas; tidak ada dependensi di luar `requests`."""
|
||||
|
||||
def __init__(self, url: str, user: str, password: str | None, verify: bool) -> None:
|
||||
self.url = url.rstrip("/")
|
||||
self.session = requests.Session()
|
||||
self.session.verify = verify
|
||||
self.session.auth = (user, password)
|
||||
self.session.headers.update({"X-Trino-User": user, "Content-Type": "text/plain"})
|
||||
|
||||
def pages(self, sql: str):
|
||||
"""Iterasi (columns, rows) per halaman; raise bila Trino melaporkan error."""
|
||||
response = self.session.post(f"{self.url}/v1/statement", data=sql, timeout=300)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
for _ in range(10_000): # pengaman halaman, bukan batas logis
|
||||
error = payload.get("error")
|
||||
if error:
|
||||
raise RuntimeError(f"trino error: {error.get('message', error)}")
|
||||
columns = [col.get("name") for col in payload.get("columns", [])]
|
||||
yield columns, payload.get("data", [])
|
||||
next_uri = payload.get("nextUri")
|
||||
if not next_uri:
|
||||
return
|
||||
payload = self.session.get(next_uri, timeout=300).json()
|
||||
raise RuntimeError("trino: terlalu banyak halaman tanpa akhir")
|
||||
|
||||
def execute(self, sql: str) -> None:
|
||||
"""Jalankan statement tanpa peduli baris (mis. CTAS)."""
|
||||
for _ in self.pages(sql):
|
||||
pass
|
||||
|
||||
def scalar(self, sql: str):
|
||||
"""Nilai pertama baris pertama."""
|
||||
for _columns, rows in self.pages(sql):
|
||||
if rows:
|
||||
return rows[0][0]
|
||||
return None
|
||||
|
||||
def checksum(self, table: str, max_rows: int) -> str:
|
||||
"""sha256 deterministik atas header + baris (JSON ringkas), opsional dibatasi."""
|
||||
digest = hashlib.sha256()
|
||||
header_written = False
|
||||
seen = 0
|
||||
for columns, rows in self.pages(f"SELECT * FROM {table}"):
|
||||
if not header_written and columns:
|
||||
digest.update((",".join(columns) + "\n").encode("utf-8"))
|
||||
header_written = True
|
||||
for row in rows:
|
||||
if max_rows and seen >= max_rows:
|
||||
return f"sha256:{digest.hexdigest()} (bounded:{max_rows})"
|
||||
digest.update(
|
||||
(json.dumps(row, separators=(",", ":"), default=str) + "\n").encode("utf-8")
|
||||
)
|
||||
seen += 1
|
||||
return f"sha256:{digest.hexdigest()}"
|
||||
|
||||
|
||||
def trino_client() -> Trino:
|
||||
password = setting("trino_password")
|
||||
if not password:
|
||||
raise RuntimeError("trino_password belum diatur (Airflow Variable atau TRINO_PASSWORD)")
|
||||
return Trino(
|
||||
setting("trino_url", DEFAULT_TRINO_URL) or DEFAULT_TRINO_URL,
|
||||
setting("trino_user", DEFAULT_TRINO_USER) or DEFAULT_TRINO_USER,
|
||||
password,
|
||||
str(setting("gold_tls_verify", "false")).lower() in {"1", "true", "yes"},
|
||||
)
|
||||
|
||||
|
||||
# ── DAG ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dag(
|
||||
dag_id="sitomat_gold_build_prod",
|
||||
schedule=None,
|
||||
catchup=False,
|
||||
tags=["sitomat", "f16", "gold"],
|
||||
doc_md=__doc__,
|
||||
)
|
||||
def sitomat_gold_build_prod():
|
||||
@task
|
||||
def build_gold() -> dict:
|
||||
"""CTAS gold dari silverRefs; kembalikan identitas artefak."""
|
||||
conf = get_current_context()["dag_run"].conf or {}
|
||||
build_id = str(conf.get("dataProductBuildId", "")).strip()
|
||||
# Run DIAGNOSTIK (`bun run check:airflow`) tidak punya build nyata.
|
||||
diagnostic = not build_id or build_id == "airflow-check"
|
||||
|
||||
source_count = _as_int(conf.get("sourceCount"))
|
||||
sources = [
|
||||
qualified(raw)
|
||||
for raw in str(conf.get("silverRefs") or "").split(",")
|
||||
if raw.strip()
|
||||
]
|
||||
missing = max(0, source_count - len(sources))
|
||||
explicit = setting("gold_sql")
|
||||
|
||||
# Run diagnostik tanpa sumber: tidak ada yang dibangun, tidak ada callback.
|
||||
if diagnostic and not sources and not explicit:
|
||||
print("[sitomat] run diagnostik tanpa sumber — tidak membangun, tidak callback")
|
||||
return {"built": False}
|
||||
|
||||
# PRODUK PARSIAL DITOLAK. Sumber yang belum punya `artifact_ref` TIDAK
|
||||
# boleh diam-diam dihilangkan (itu menghasilkan gold yang salah). Untuk
|
||||
# build NYATA kita GAGAL di sini, sehingga task `report_failure`
|
||||
# melaporkannya ke sitomat — bukan menggantung `building` selamanya.
|
||||
if missing > 0 and not explicit:
|
||||
raise ValueError(
|
||||
f"build {build_id}: {missing} dari {source_count} sumber tidak punya "
|
||||
"artifact_ref — menolak membangun produk parsial"
|
||||
)
|
||||
|
||||
target, sql = target_and_sql(conf, sources)
|
||||
client = trino_client()
|
||||
print(f"[sitomat] membangun {target} dari {len(sources)} sumber")
|
||||
client.execute(sql)
|
||||
|
||||
row_count = int(client.scalar(f"SELECT count(*) FROM {target}") or 0)
|
||||
checksum = client.checksum(target, int(setting("checksum_max_rows", str(DEFAULT_CHECKSUM_MAX_ROWS))))
|
||||
print(f"[sitomat] selesai: {target} rows={row_count} checksum={checksum}")
|
||||
return {
|
||||
"built": True,
|
||||
"goldRef": target,
|
||||
"rowCount": row_count,
|
||||
"checksum": checksum,
|
||||
"buildId": str(conf.get("dataProductBuildId", "")),
|
||||
}
|
||||
|
||||
@task
|
||||
def register_gold(build: dict) -> None:
|
||||
"""Laporkan artefak ke sitomat (callback `register_gold`)."""
|
||||
if not build.get("built"):
|
||||
return
|
||||
base = (setting("sitomat_base_url", DEFAULT_BASE_URL) or DEFAULT_BASE_URL).rstrip("/")
|
||||
api_key = setting("sitomat_api_key")
|
||||
if not api_key:
|
||||
raise RuntimeError("sitomat_api_key belum diatur (Airflow Variable atau SITOMAT_API_KEY)")
|
||||
url = f"{base}/data-products/builds/{build['buildId']}/gold"
|
||||
response = requests.post(
|
||||
url,
|
||||
json={
|
||||
"goldRef": build["goldRef"],
|
||||
"checksum": build["checksum"],
|
||||
"rowCount": build["rowCount"],
|
||||
},
|
||||
headers={"X-API-Key": api_key},
|
||||
timeout=60,
|
||||
)
|
||||
if response.status_code >= 300:
|
||||
raise RuntimeError(
|
||||
f"callback sitomat gagal: HTTP {response.status_code} {response.text[:300]}"
|
||||
)
|
||||
print(f"[sitomat] callback OK ({response.status_code}): {response.text[:200]}")
|
||||
|
||||
@task(trigger_rule="one_failed")
|
||||
def report_failure() -> None:
|
||||
"""Laporkan kegagalan ke sitomat agar build TIDAK menggantung `building`.
|
||||
|
||||
Best-effort: kegagalan pelaporan tidak boleh menutupi error asli DAG.
|
||||
"""
|
||||
context = get_current_context()
|
||||
conf = context["dag_run"].conf or {}
|
||||
build_id = str(conf.get("dataProductBuildId", "")).strip()
|
||||
# Run diagnostik (`bun run check:airflow`) tidak punya build nyata.
|
||||
if not build_id or build_id == "airflow-check":
|
||||
print("[sitomat] tidak ada build nyata untuk dilaporkan (mode konektivitas)")
|
||||
return
|
||||
base = (setting("sitomat_base_url", DEFAULT_BASE_URL) or DEFAULT_BASE_URL).rstrip("/")
|
||||
api_key = setting("sitomat_api_key")
|
||||
if not api_key:
|
||||
print("[sitomat] sitomat_api_key tidak diatur — kegagalan TIDAK dilaporkan")
|
||||
return
|
||||
cause = f"DAG-3 sitomat_gold_build_prod failed (run {context['dag_run'].run_id})"
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{base}/data-products/builds/{build_id}/fail",
|
||||
json={"cause": cause},
|
||||
headers={"X-API-Key": api_key},
|
||||
timeout=60,
|
||||
)
|
||||
print(f"[sitomat] failure callback -> HTTP {response.status_code}")
|
||||
except Exception as error: # noqa: BLE001 - never mask the original failure
|
||||
print(f"[sitomat] failure callback GAGAL (diabaikan): {error}")
|
||||
|
||||
built = build_gold()
|
||||
registered = register_gold(built)
|
||||
# `one_failed` + kedua upstream: lapor bila SALAH SATU task gagal.
|
||||
[built, registered] >> report_failure()
|
||||
|
||||
|
||||
sitomat_gold_build_prod()
|
||||
@@ -0,0 +1,213 @@
|
||||
"""
|
||||
# =============================================================================
|
||||
# eBdesk SiCABE — Contoh DAG Pipeline End-to-End
|
||||
#
|
||||
# Referensi arsitektur:
|
||||
# bab 5.4 (Ingestion → Processing: Pipeline Integration)
|
||||
# bab 8.1 (Skenario End-to-End: Batch Harian)
|
||||
# bab 8.2 (Skenario End-to-End: CDC Real-Time)
|
||||
# =============================================================================
|
||||
|
||||
Alur:
|
||||
1. [Ingestion] Sling tarik data dari PostgreSQL → Bronze (Iceberg via Nessie)
|
||||
2. [Processing] dbt via Trino: Bronze → Silver → Gold
|
||||
3. [Governance] Great Expectations validasi output pipeline DI ICEBERG
|
||||
|
||||
Prerequisites (dibuat oleh bootstrap.sh):
|
||||
- Airflow connection: minio_s3 (object storage)
|
||||
- Sling env: ~/.sling/env.yaml (koneksi POSTGRES_DB + ICEBERG)
|
||||
- dbt project: /opt/airflow/dbt/sicabe/ (bind-mount dari config/airflow/dbt) dengan model bronze_to_silver, silver_to_gold, incremental
|
||||
- Tabel sumber: public.orders di database sicabe_data (PostgreSQL)
|
||||
- Kredensial Trino service account: env DBT_TRINO_USER / DBT_TRINO_PASSWORD
|
||||
|
||||
Catatan penting (jangan diubah tanpa alasan):
|
||||
- Task dbt memakai `env=DBT_TASK_ENV, append_env=True`. Tanpa `append_env=True`,
|
||||
`env=` MENGGANTI seluruh environment sehingga PATH & HOME hilang dan `dbt`
|
||||
(terpasang di ~/.local/bin) tidak ditemukan: "dbt: command not found".
|
||||
- Validator Great Expectations membaca tabel Iceberg LEWAT TRINO (HTTPS 8443,
|
||||
service account), bukan tabel sumber di PostgreSQL — supaya hasil "LULUS"
|
||||
benar-benar membuktikan output pipeline (bronze/gold), bukan data mentah.
|
||||
GE tidak punya datasource Trino → dipakai Pandas datasource dari hasil query.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
import os
|
||||
|
||||
from airflow import DAG
|
||||
from airflow.providers.standard.operators.bash import BashOperator
|
||||
from airflow.providers.standard.operators.python import PythonOperator
|
||||
|
||||
|
||||
def _run_ge_to_om(table):
|
||||
"""Jalankan validasi Great Expectations DAN kirim hasilnya ke OpenMetadata.
|
||||
|
||||
Runner adalah SATU sumber logika validasi (dipakai juga di luar DAG) dan
|
||||
memakai action resmi OpenMetadata untuk GE 1.x, sehingga hasilnya muncul
|
||||
sebagai Data Quality Test Suite/Test Case di OpenMetadata — TERMASUK ketika
|
||||
GAGAL. Sebelumnya hasil GE hanya berhenti di log Airflow.
|
||||
|
||||
Fail-fast tetap dipertahankan: exit code non-nol → task FAILED.
|
||||
"""
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
script = os.getenv(
|
||||
"GE_OM_RUNNER", "/opt/airflow/great_expectations/validate_to_openmetadata.py"
|
||||
)
|
||||
proc = subprocess.run(
|
||||
[sys.executable, script, "--table", table], capture_output=True, text=True
|
||||
)
|
||||
for stream in (proc.stdout, proc.stderr):
|
||||
if stream:
|
||||
print(stream[-2000:])
|
||||
if proc.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"Data quality check GAGAL untuk {table} (hasil tetap dikirim ke OpenMetadata)"
|
||||
)
|
||||
|
||||
|
||||
def validate_gold_data():
|
||||
"""Validasi GOLD (output akhir pipeline) — `gold.orders_gold` → hasil ke OM."""
|
||||
_run_ge_to_om("gold.orders_gold")
|
||||
|
||||
|
||||
def validate_bronze_data():
|
||||
"""Validasi BRONZE (hasil ingestion Sling) — `bronze.orders` → hasil ke OM."""
|
||||
_run_ge_to_om("bronze.orders")
|
||||
|
||||
|
||||
# Env untuk task dbt: autentikasi Trino via SERVICE ACCOUNT (HTTP Basic) langsung
|
||||
# ke HTTPS 8443 (lihat config/airflow/dbt/profiles.yml). Password dari env
|
||||
# DBT_TRINO_PASSWORD (di-set docker-compose untuk service Airflow).
|
||||
DBT_TASK_ENV = {
|
||||
"DBT_TRINO_USER": os.getenv("DBT_TRINO_USER", "svc_dbt"),
|
||||
"DBT_TRINO_PASSWORD": os.getenv("DBT_TRINO_PASSWORD", ""),
|
||||
}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Default Arguments
|
||||
# =============================================================================
|
||||
default_args = {
|
||||
"owner": "engineer",
|
||||
"depends_on_past": False,
|
||||
"email_on_failure": True,
|
||||
"email_on_retry": False,
|
||||
"retries": 2,
|
||||
"retry_delay": timedelta(minutes=5),
|
||||
"execution_timeout": timedelta(hours=2),
|
||||
}
|
||||
|
||||
# =============================================================================
|
||||
# DAG — Batch Harian (bab 8.1)
|
||||
# =============================================================================
|
||||
BATCH_DAG_ID = "sicabe_batch_daily"
|
||||
BATCH_SCHEDULE = "0 2 * * *" # Setiap hari jam 02:00
|
||||
|
||||
with DAG(
|
||||
dag_id=BATCH_DAG_ID,
|
||||
default_args=default_args,
|
||||
description="SiCABE: Batch harian — Ingestion → Processing → Governance",
|
||||
schedule=BATCH_SCHEDULE,
|
||||
start_date=datetime(2026, 1, 1),
|
||||
catchup=False,
|
||||
tags=["sicabe", "batch", "bronze", "silver", "gold"],
|
||||
) as dag:
|
||||
|
||||
# ─── Step 1: Ingestion — Tarik data ke Bronze (Iceberg) ──────────────
|
||||
ingest_to_bronze = BashOperator(
|
||||
task_id="ingest_to_bronze",
|
||||
bash_command=(
|
||||
'sling run --src-conn POSTGRES_DB '
|
||||
'--src-stream public.orders '
|
||||
'--tgt-conn ICEBERG '
|
||||
'--tgt-object bronze.orders '
|
||||
'--mode full-refresh '
|
||||
'--stdout'
|
||||
),
|
||||
)
|
||||
|
||||
# ─── Step 2: Processing — Transform Bronze → Silver ──────────────────
|
||||
bronze_to_silver = BashOperator(
|
||||
task_id="bronze_to_silver",
|
||||
bash_command=(
|
||||
'cd /opt/airflow/dbt/sicabe && '
|
||||
'export DBT_PROFILES_DIR=/opt/airflow/dbt && '
|
||||
'dbt run --select bronze_to_silver --profile trino --target prod'
|
||||
),
|
||||
env=DBT_TASK_ENV,
|
||||
append_env=True,
|
||||
)
|
||||
|
||||
# ─── Step 3: Processing — Transform Silver → Gold ────────────────────
|
||||
silver_to_gold = BashOperator(
|
||||
task_id="silver_to_gold",
|
||||
bash_command=(
|
||||
'cd /opt/airflow/dbt/sicabe && '
|
||||
'export DBT_PROFILES_DIR=/opt/airflow/dbt && '
|
||||
'dbt run --select silver_to_gold --profile trino --target prod'
|
||||
),
|
||||
env=DBT_TASK_ENV,
|
||||
append_env=True,
|
||||
)
|
||||
|
||||
# ─── Step 4: Governance — Validasi GOLD di Iceberg ───────────────────
|
||||
validate_gold = PythonOperator(
|
||||
task_id="validate_gold",
|
||||
python_callable=validate_gold_data,
|
||||
)
|
||||
|
||||
# ─── Dependency Chain ────────────────────────────────────────────────
|
||||
ingest_to_bronze >> bronze_to_silver >> silver_to_gold >> validate_gold
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DAG — CDC Real-Time (bab 8.2)
|
||||
# =============================================================================
|
||||
CDC_DAG_ID = "sicabe_cdc_realtime"
|
||||
CDC_SCHEDULE = "*/15 * * * *" # Setiap 15 menit
|
||||
|
||||
with DAG(
|
||||
dag_id=CDC_DAG_ID,
|
||||
default_args=default_args,
|
||||
description="SiCABE: CDC real-time — Incremental ingestion + processing",
|
||||
schedule=CDC_SCHEDULE,
|
||||
start_date=datetime(2026, 1, 1),
|
||||
catchup=False,
|
||||
tags=["sicabe", "cdc", "incremental", "realtime"],
|
||||
) as dag:
|
||||
|
||||
# ─── Step 1: CDC Ingestion via Sling ─────────────────────────────────
|
||||
cdc_ingest = BashOperator(
|
||||
task_id="cdc_ingest",
|
||||
bash_command=(
|
||||
'sling run --src-conn POSTGRES_DB '
|
||||
'--src-stream public.orders '
|
||||
'--tgt-conn ICEBERG '
|
||||
'--tgt-object bronze.orders '
|
||||
'--mode incremental '
|
||||
'--primary-key id '
|
||||
'--update-key updated_at '
|
||||
'--stdout'
|
||||
),
|
||||
)
|
||||
|
||||
# ─── Step 2: dbt incremental update ──────────────────────────────────
|
||||
dbt_incremental = BashOperator(
|
||||
task_id="dbt_incremental",
|
||||
bash_command=(
|
||||
'cd /opt/airflow/dbt/sicabe && '
|
||||
'export DBT_PROFILES_DIR=/opt/airflow/dbt && '
|
||||
'dbt run --select incremental --profile trino --target prod'
|
||||
),
|
||||
env=DBT_TASK_ENV,
|
||||
append_env=True,
|
||||
)
|
||||
|
||||
# ─── Step 3: Validasi BRONZE di Iceberg ──────────────────────────────
|
||||
cdc_validate = PythonOperator(
|
||||
task_id="cdc_validate",
|
||||
python_callable=validate_bronze_data,
|
||||
)
|
||||
|
||||
cdc_ingest >> dbt_incremental >> cdc_validate
|
||||
@@ -0,0 +1,58 @@
|
||||
# =============================================================================
|
||||
# SAMPLE — Lineage otomatis via operator SQL (tanpa inlets/outlets)
|
||||
# =============================================================================
|
||||
# Tujuan : membuktikan bahwa task yang memakai operator SQL memperoleh lineage
|
||||
# tabel SECARA OTOMATIS dari teks SQL, tanpa perlu menuliskan
|
||||
# inlets/outlets.
|
||||
#
|
||||
# Alur lineage yang diharapkan:
|
||||
# iceberg.bronze.orders ---> iceberg.silver.orders_clean
|
||||
#
|
||||
# Cara kerja:
|
||||
# Provider OpenLineage mem-parsing SQL task (lihat
|
||||
# airflow.providers.openlineage.utils.sql_hook_lineage) dan mengekstrak
|
||||
# tabel input/output, lalu mengirimnya ke OpenMetadata.
|
||||
#
|
||||
# Prasyarat di Airflow:
|
||||
# 1. Koneksi `trino_ol` (type trino) menunjuk ke Trino koordinator.
|
||||
# 2. Transport OpenLineage aktif + shim eventTime (lihat shim/ol_time_shim.py).
|
||||
# 3. AIRFLOW__OPENLINEAGE__EXECUTE_IN_THREAD=true (tanpa ini emisi HANG).
|
||||
#
|
||||
# Catatan: DAG ini tidak memakai inlets/outlets secara sengaja.
|
||||
# =============================================================================
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from airflow import DAG
|
||||
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
|
||||
|
||||
CONN_ID = "trino_ol"
|
||||
|
||||
CLEAN_SQL = """
|
||||
CREATE TABLE IF NOT EXISTS iceberg.silver.orders_clean AS
|
||||
SELECT
|
||||
id,
|
||||
order_date,
|
||||
komoditas,
|
||||
harga_per_kg,
|
||||
stok_kg,
|
||||
region
|
||||
FROM iceberg.bronze.orders
|
||||
WHERE id IS NOT NULL
|
||||
"""
|
||||
|
||||
with DAG(
|
||||
dag_id="sample_sql_lineage",
|
||||
description="SAMPLE: lineage otomatis via operator SQL (bronze.orders -> silver.orders_clean)",
|
||||
start_date=datetime(2026, 1, 1),
|
||||
schedule=None, # dijalankan manual
|
||||
catchup=False,
|
||||
tags=["sample", "lineage", "openlineage", "sql"],
|
||||
) as dag:
|
||||
|
||||
# Lineage bronze.orders -> silver.orders_clean diekstrak otomatis dari SQL ini.
|
||||
clean_orders = SQLExecuteQueryOperator(
|
||||
task_id="clean_orders",
|
||||
conn_id=CONN_ID,
|
||||
sql=CLEAN_SQL,
|
||||
)
|
||||
@@ -0,0 +1,9 @@
|
||||
"""
|
||||
This file has been generated from dag_runner.j2
|
||||
"""
|
||||
from airflow import DAG
|
||||
from openmetadata_managed_apis.workflows import workflow_factory
|
||||
|
||||
workflow = workflow_factory.WorkflowFactory.create("/opt/airflow/dag_generated_configs/sicabe_dq_gold_native.json")
|
||||
workflow.generate_dag(globals())
|
||||
dag = workflow.get_dag()
|
||||
Reference in New Issue
Block a user