dlt في 2026: بناء خطوط أنابيب ETL بايثون مع تطور المخطط والتحميل التزايدي

آخر تحديث: 29 يونيو 2026

dlt (data load tool) هي مكتبة بايثون مفتوحة المصدر تحوّل أي مصدر بيانات منظم أو شبه منظم إلى جداول مُطبَّعة في وجهتك المفضلة (DuckDB، BigQuery، Snowflake، Postgres) دون أن تكتب سطر SQL واحد للـ DDL، مع تطور تلقائي للمخطط، تحميل تزايدي مدمج، وحالة قابلة للاستعادة. في 2026 أصبحت dlt الخيار العملي عندما تريد بساطة Singer/Meltano دون عبء العملاء الخارجيين، وقابلية إنتاج Airbyte دون الحاجة لخادم. هذا الدليل يأخذك من أول pipeline إلى نمط إنتاجي مع اختبارات وتكامل dbt.

• dlt تتعامل مع DDL، التطبيع، وتطور المخطط تلقائياً، وأنت تركّز على استخراج البيانات فقط.
• التحميل التزايدي يعتمد على state مستمرة بين عمليات التشغيل، فلا تحتاج لجدول تتبع منفصل.
• تطور المخطط له ثلاثة أوضاع: evolve، freeze، وdiscard_row، تختار حسب صرامة العقد المطلوب.
• تتكامل dlt أصلياً مع dbt عبر dlt.dbt لتشغيل نماذج التحويل بعد التحميل في نفس المرحلة.
• اختبارات الـ pipeline ليست اختيارية: استخدم pytest مع وجهة DuckDB في الذاكرة للتحقق من المخطط وعدد الصفوف.
• الإصدار 1.x مستقر منذ أواخر 2024، وأضافت نسخ 2026 دعماً أصلياً لـ Apache Iceberg و Delta Lake كوجهات.

ما هي dlt ولماذا تهم مهندس البيانات في 2026؟

بصراحة، أمضيت سنوات أبني خطوط أنابيب بـ Singer taps و Meltano، وكان معظم وقتي يذهب لإصلاح schemas تنكسر بصمت عند الثالثة فجراً. dlt تعالج المشكلة بنهج مختلف: بدلاً من بروتوكول خارجي ومجتمع من العملاء بجودة متفاوتة، تكتب دالة بايثون عادية تُرجع سجلات (dict أو generator)، و dlt تستنتج المخطط، تنشئ الجداول، تطبّع JSON المتداخل، وتحمّل البيانات إلى وجهتك.

الفكرة المركزية بسيطة: أنت مسؤول فقط عن الاستخراج. كل ما يلي ذلك (اشتقاق المخطط، تطبيع البيانات، التحميل الذري، تتبع الحالة، تطور المخطط) تتعامل معه المكتبة. في الإنتاج هذا يعني أن مراجعة الكود تركز على منطق الأعمال بدل عبارات CREATE TABLE الطويلة. الإصدار 1.0 صدر في سبتمبر 2024، وأضافت نسخ 2026 دعماً مستقراً لـ Apache Iceberg و Delta Lake وتحسينات كبيرة في أداء التحميل المتوازي. للاطلاع على التفاصيل الكاملة راجع التوثيق الرسمي لـ dlt والمستودع على GitHub.

التثبيت وأول pipeline في خمس دقائق

أبسط طريقة للبدء هي تثبيت dlt مع موصل DuckDB. لا حساب سحابي ولا إعدادات شبكة. سنحمّل بيانات من GitHub API إلى جدول محلي:

pip install "dlt[duckdb]>=1.5.0" requests

import dlt
import requests

@dlt.resource(name="issues", write_disposition="replace")
def github_issues(repo: str = "duckdb/duckdb"):
    url = f"https://api.github.com/repos/{repo}/issues"
    params = {"state": "all", "per_page": 100}
    while url:
        r = requests.get(url, params=params, timeout=30)
        r.raise_for_status()
        yield r.json()
        url = r.links.get("next", {}).get("url")
        params = None  # next link contains them

pipeline = dlt.pipeline(
    pipeline_name="github_demo",
    destination="duckdb",
    dataset_name="github_raw",
)

load_info = pipeline.run(github_issues())
print(load_info)

عند التشغيل، dlt تستنتج المخطط من السجلات الفعلية، تنشئ جداول للحقول المسطحة، وتطبّع كل قائمة متداخلة (مثل labels أو assignees) إلى جدول فرعي مع عمود _dlt_parent_id يربطه بالأصل. تستطيع الاستعلام مباشرة عبر pipeline.dataset().issues.df() للحصول على pandas DataFrame.

كيف يعمل التحميل التزايدي في dlt؟

التحميل التزايدي هو المكان الذي تتفوق فيه dlt على الأدوات اليدوية. بدلاً من إدارة جدول watermark بنفسك، تستخدم dlt.sources.incremental الذي يحفظ آخر قيمة معالجة في state الـ pipeline تلقائياً، ويعيد استخدامها في التشغيل التالي:

from datetime import datetime, timezone

@dlt.resource(
    name="issues",
    write_disposition="merge",
    primary_key="id",
)
def github_issues_incremental(
    repo: str,
    updated_at = dlt.sources.incremental(
        "updated_at",
        initial_value="2024-01-01T00:00:00Z",
    ),
):
    url = f"https://api.github.com/repos/{repo}/issues"
    params = {
        "state": "all",
        "per_page": 100,
        "since": updated_at.last_value,  # من الحالة المحفوظة
        "sort": "updated",
        "direction": "asc",
    }
    while url:
        r = requests.get(url, params=params, timeout=30)
        r.raise_for_status()
        yield r.json()
        url = r.links.get("next", {}).get("url")
        params = None

عند التشغيل الأول تأخذ القيمة الأولية. عند التشغيل التالي تستخدم آخر updated_at رأته. مع write_disposition="merge" وprimary_key="id" تقوم dlt بـ MERGE الصفوف الجديدة فوق الموجودة، فلا تكرار حتى لو عاد نفس السجل. الحالة محفوظة داخل وجهتك في جدول _dlt_pipeline_state، مما يعني أن أي عامل (worker) في أي مكان يستطيع استئناف الـ pipeline.

أنماط write_disposition الأربعة

• append: يضيف الصفوف الجديدة فقط، مناسب لجداول الأحداث (event tables).
• replace: يحذف الجدول ويعيد إنشاءه، للأبعاد الصغيرة (small dims) التي تتغير كلياً.
• merge: UPSERT حقيقي يعتمد على primary_key، وهو الخيار الأساسي للجداول التشغيلية.
• scd2 (جديد في 2026): يحفظ السجل التاريخي مع _dlt_valid_from و_dlt_valid_to، لـ Slowly Changing Dimensions type 2.

تطور المخطط: evolve مقابل freeze

الـ schema drift هي السبب رقم واحد لانكسار خطوط الأنابيب بصمت. dlt تعطيك ثلاثة خيارات صريحة عبر schema_contract:

الوضع	عند ظهور عمود جديد	عند تغيير نوع عمود	الاستخدام النموذجي
evolve	إضافة العمود تلقائياً	توسيع النوع إن أمكن	البحث والاستكشاف، landing zone
freeze	فشل الـ pipeline	فشل الـ pipeline	طبقات gold التي تغذي تقارير المالية
discard_row	تجاهل الصف مع تحذير	تجاهل الصف	مصادر صاخبة حيث الـ availability أهم من الاكتمال
discard_value	إسقاط القيمة فقط	إسقاط القيمة فقط	عندما يكون الصف مفيداً حتى مع حقول مفقودة

@dlt.resource(
    name="orders",
    schema_contract={"tables": "freeze", "columns": "evolve", "data_type": "freeze"},
)
def orders():
    ...

قاعدة عملية من تجربتي: استخدم evolve في طبقة raw، وfreeze في طبقات silver/gold. هذا يعطيك مرونة الاستيعاب وصرامة العقد عند نقطة تماس المستهلكين. للمزيد عن أنماط التحقق من البيانات في بايثون، راجع دليلنا حول التحقق من صحة البيانات في pandas باستخدام Pandera، فهو يكمل dlt جيداً على جانب التحويل.

dlt مقابل Airbyte و Singer/Meltano

السؤال الذي أُسأله أسبوعياً: هل أستبدل Airbyte بـ dlt؟ الإجابة تعتمد على من يدير الـ pipeline.

المعيار	dlt	Airbyte	Singer/Meltano
طريقة التعريف	كود بايثون	UI + YAML	YAML + taps
تطور المخطط	مدمج وقابل للتكوين	مدعوم	يدوي عادة
التحميل التزايدي	state تلقائية	cursor field	state file يدوي
البنية التحتية	صفر، pip فقط	Docker / حساب سحابي	Python + meltano server
المراقبة	logs + load_info	UI كامل	محدودة
منحنى التعلم	منخفض لمطور بايثون	منخفض لغير المطور	متوسط

توصيتي بعد ثلاث migrations فعلية: إذا كان فريقك يكتب بايثون يومياً ويفضّل أن يعيش الـ pipeline في git مع PRs ومراجعات، فـ dlt هي الفائز. إذا كنت تحتاج موصلات SaaS غير شائعة (Salesforce، NetSuite) بمنبع جاهز و GUI، فـ Airbyte أو Fivetran أفضل. Singer/Meltano أصبحا في وضع صيانة عملياً، ولا أبني عليهما مشاريع جديدة في 2026.

الوجهات في 2026: DuckDB، BigQuery، Iceberg

dlt تدعم أكثر من 20 وجهة بنفس واجهة destination=. التغيير الأبرز في 2026 هو نضوج موصلات lakehouse:

# Iceberg على S3 (جديد ومستقر منذ 0.5.x)
pipeline = dlt.pipeline(
    pipeline_name="events_lake",
    destination=dlt.destinations.filesystem(
        bucket_url="s3://my-lake/iceberg",
    ),
    dataset_name="events",
    progress="log",
)
load_info = pipeline.run(
    events_resource(),
    table_format="iceberg",  # كتابة جداول Iceberg أصلية
)

هذا يعني أنك تستطيع كتابة جداول Iceberg مباشرة من dlt دون Spark، و Athena، Trino، أو DuckDB يقرؤونها فوراً. للتحميل إلى DuckDB محلياً للتحليل، راجع دليلنا حول DuckDB في بايثون 2026 الذي يكمل dlt كطبقة استعلام. ولأنماط التوافق بين pandas و Polars بعد التحميل، اطّلع أيضاً على Narwhals 2026 للتوافق بين pandas و Polars.

الاعتمادات الإضافية لكل وجهة

pip install "dlt[bigquery]"      # BigQuery
pip install "dlt[snowflake]"     # Snowflake
pip install "dlt[postgres]"      # Postgres / Redshift
pip install "dlt[filesystem,s3]" # S3 / GCS / Azure للـ lake
pip install "dlt[athena]"        # Athena + Iceberg

التكامل مع dbt في نفس الـ pipeline

كمدمن dbt، الميزة التي جعلتني أحوّل فريقي بالكامل إلى dlt هي dlt.dbt. تستطيع تشغيل مشروع dbt كاملاً مباشرة بعد التحميل في نفس عملية Python، فلا حاجة لـ Airflow لربط المرحلتين:

import dlt
from dlt.helpers.dbt import package as dbt_package

pipeline = dlt.pipeline(
    pipeline_name="orders_e2e",
    destination="duckdb",
    dataset_name="staging",
)

# 1) استخراج وتحميل
pipeline.run(orders_resource())

# 2) تشغيل dbt مباشرة بعد التحميل
dbt = dbt_package(
    pipeline,
    package_location="./dbt_project",
    venv=None,  # استخدم البيئة الحالية
)
models = dbt.run_all()
for m in models:
    print(m.model_name, m.status, m.message)

الذكاء هنا: dlt تمرّر credentials الوجهة و schema الحالية تلقائياً إلى dbt عبر profiles.yml مُنشأ ديناميكياً. اختبارات dbt تُشغَّل بـ dbt.test()، وإذا فشل اختبار فإن الـ pipeline كاملاً يفشل، وهو ما تريده في pipeline موجه نحو العقود.

اختبار خطوط الأنابيب، لأنها ستنكسر

هذا الجزء الذي يهملونه في 90% من الفيديوهات على يوتيوب، وهو السبب رقم واحد لمكالمات الـ on-call. اختبارات الـ pipeline في dlt سهلة بشكل صادم لأن DuckDB في الذاكرة وجهة كاملة:

# test_pipeline.py
import dlt
import pytest

def test_schema_is_stable():
    pipeline = dlt.pipeline(
        pipeline_name="test_run",
        destination="duckdb",
        dataset_name="test_ds",
        dev_mode=True,  # وجهة جديدة لكل اختبار
    )
    sample = [
        {"id": 1, "user": {"name": "ali"}, "tags": ["a", "b"]},
        {"id": 2, "user": {"name": "sara"}, "tags": ["c"]},
    ]
    info = pipeline.run(sample, table_name="events")
    assert info.has_failed_jobs is False

    schema = pipeline.default_schema
    cols = schema.tables["events"]["columns"]
    assert "id" in cols
    assert cols["id"]["data_type"] == "bigint"
    # التطبيع: tags ينبغي أن تصبح جدولاً فرعياً
    assert "events__tags" in schema.tables

def test_row_count_after_run():
    pipeline = dlt.pipeline(
        pipeline_name="test_count",
        destination="duckdb",
        dev_mode=True,
    )
    pipeline.run([{"id": i} for i in range(100)], table_name="nums")
    with pipeline.sql_client() as c:
        rows = c.execute_sql("SELECT COUNT(*) FROM nums")[0][0]
    assert rows == 100

أقل ما يجب أن يحتويه كل مشروع dlt في الإنتاج: اختبار يتحقق من المخطط (لمنع تطور غير مقصود)، اختبار يتحقق من has_failed_jobs، واختبار يتحقق من عدد صفوف معقول. شغّلها في CI قبل كل merge، فكلفة الـ CI لـ pipeline اختباري بـ DuckDB لا تتعدى مللي ثوانٍ.

الترحيل (Backfill) وأنماط الإنتاج

أكثر سؤال يأتيني: "كيف أعمل backfill لشهر مفقود؟". في dlt، الـ state هي مصدر الحقيقة لآخر قيمة معالجة. لـ backfill نظيف:

1. لا تعدّل الحالة يدوياً. هذا يكسر تتبع الـ run.
2. أنشئ مثيل pipeline منفصل باسم pipeline_name="orders_backfill_2026_01" مع نطاق ثابت داخل المورد (updated_at_min، updated_at_max) ووجهة منفصلة أو schema منفصل.
3. اكتب إلى جدول staging، تحقق من الصفوف، ثم MERGE إلى الجدول الإنتاجي بعملية SQL واحدة محكمة.
4. وثّق الـ backfill في git commit بتفاصيل النطاق والسبب.

لتشغيل المهام في الإنتاج، dlt تعمل داخل أي scheduler: cron مباشرة، GitHub Actions، Airflow (هناك DltOperator)، Dagster (asset مدمج)، أو Prefect. الميزة أن dlt هي مكتبة بايثون عادية، فهي لا تفرض runtime معيناً. لتقديم النتائج عبر API بعد التحميل، يعمل دليلنا حول نشر نماذج scikit-learn مع FastAPI كنمط مكمل: dlt تجلب، و FastAPI تقدم.

الأسئلة الشائعة

هل dlt بديل عن Airflow؟

لا. dlt تتعامل مع الـ EL (الاستخراج والتحميل) فقط، بينما Airflow هو scheduler يدير اعتماديات بين مهام مختلفة. الاثنان يكملان بعضهما: تكتب الـ pipeline بـ dlt وتجدوله بـ Airflow أو Dagster أو حتى cron بسيط.

ما الفرق بين @dlt.resource و @dlt.source؟

resource يمثل جدولاً واحداً (دالة تُرجع سجلات). source يجمع عدة resources منطقياً (مثل "كل جداول Stripe") مع إعدادات مشتركة وعلاقات. ابدأ بـ resource؛ ارفع إلى source عندما تشاركها بين مشاريع.

كيف أعالج البيانات الكبيرة جداً (TB+) بـ dlt؟

استخدم chunk_size في المورد لتقسيم الـ generator، فعّل parallelism في إعدادات الـ runtime، واكتب إلى وجهة filesystem بصيغة Parquet مع table_format="iceberg" أو "delta". dlt تستخدم PyArrow داخلياً لتجنّب نسخ البيانات في الذاكرة.

هل dlt آمنة للبيانات الحساسة؟

نعم. الـ credentials تُقرأ من متغيرات البيئة، أو secrets.toml، أو vault خارجي (HashiCorp Vault، AWS Secrets Manager). dlt لا تُسجّل قيم البيانات الفعلية افتراضياً، فقط schema والإحصائيات. لتقنين أعمق، استخدم column hints مع "data_type": "text" ومعالج تشفير قبل التحميل.

هل أحتاج معرفة بـ SQL لاستخدام dlt؟

للاستيعاب الأساسي: لا. dlt تُنشئ الجداول وتحمّل البيانات دون كتابة SQL. لكن لمراجعة المخطط الناتج، تحسين MERGE، أو كتابة نماذج dbt لاحقة، تحتاج SQL متوسط. dlt تُخفض عبء SQL لكنها لا تُلغيه.