نموذج تنفيذ DataStore

فهم التقييم الكسول، ومحفزات التنفيذ، والتخزين المؤقت في DataStore

يُعد فهم نموذج التقييم الكسول في DataStore مفتاحًا لاستخدامه بفعالية وتحقيق أفضل أداء.

​

التقييم الكسول

يستخدم DataStore التقييم الكسول — إذ لا تُنفَّذ العمليات فورًا، بل تُسجَّل وتُترجم إلى استعلامات SQL مُحسّنة. ولا يتم التنفيذ إلا عند الحاجة الفعلية إلى النتائج.

​

مثال: التقييم الكسول مقابل التقييم الفوري

from pathlib import Path
Path("sales.csv").write_text("""\
region,product,category,amount,quantity,price,date,order_id
East,Widget,Electronics,5200,10,120,2024-01-15,1001
West,Gadget,Electronics,800,5,160,2024-02-20,1002
East,Gizmo,Home,6500,3,100,2024-03-10,1003
North,Widget,Electronics,4500,6,150,2024-06-18,1004
West,Gadget,Electronics,2000,8,250,2024-09-14,1005
""")

from chdb import datastore as pd

ds = pd.read_csv("sales.csv")

# These operations are NOT executed yet
result = (ds
    .filter(ds['amount'] > 1000)    # Recorded, not executed
    .select('region', 'amount')      # Recorded, not executed
    .groupby('region')               # Recorded, not executed
    .agg({'amount': 'sum'})          # Recorded, not executed
    .sort('sum', ascending=False)    # Recorded, not executed
)

# Still no execution - just building the query plan
print(result.to_sql())
# SELECT region, SUM(amount) AS sum
# FROM file('sales.csv', 'CSVWithNames')
# WHERE amount > 1000
# GROUP BY region
# ORDER BY sum DESC

# NOW execution happens
df = result.to_df()  # <-- Triggers execution

​

فوائد التقييم الكسول

1. تحسين الاستعلام: تتحول عمليات متعددة إلى استعلام SQL واحد مُحسَّن
2. تمرير عوامل التصفية إلى المصدر: تُطبَّق عوامل التصفية على مستوى مصدر البيانات
3. تقليم الأعمدة: لا تُقرأ إلا الأعمدة المطلوبة
4. القرارات المؤجلة: يمكن اختيار محرك التنفيذ في وقت التشغيل
5. فحص الخطة: يمكنك عرض الاستعلام أو تصحيحه قبل تنفيذه

---

​

مشغّلات التنفيذ

يبدأ التنفيذ تلقائيًا عند الحاجة إلى القيم الفعلية:

​

المشغّلات التلقائية

المشغّل	مثال	الوصف
print() / repr()	print(ds)	عرض النتائج
len()	len(ds)	الحصول على عدد الصفوف
.columns	ds.columns	الحصول على أسماء الأعمدة
.dtypes	ds.dtypes	الحصول على أنواع الأعمدة
.shape	ds.shape	الحصول على الأبعاد
.index	ds.index	الحصول على فهرس الصفوف
.values	ds.values	الحصول على مصفوفة NumPy
التكرار	for row in ds	التكرار على الصفوف
to_df()	ds.to_df()	التحويل إلى pandas
to_pandas()	ds.to_pandas()	اسم بديل لـ to_df
to_dict()	ds.to_dict()	التحويل إلى قاموس
to_numpy()	ds.to_numpy()	التحويل إلى مصفوفة
.equals()	ds.equals(other)	مقارنة كائنات DataStore

أمثلة:

# All these trigger execution
print(ds)              # Display
len(ds)                # 1000
ds.columns             # Index(['name', 'age', 'city'])
ds.shape               # (1000, 3)
list(ds)               # List of values
ds.to_df()             # pandas DataFrame

​

العمليات التي تظل مؤجّلة التنفيذ

العملية	القيمة المعادة	الوصف
filter()	DataStore	يضيف عبارة WHERE
select()	DataStore	يضيف تحديد الأعمدة
sort()	DataStore	يضيف ORDER BY
groupby()	LazyGroupBy	يُعِدّ GROUP BY
join()	DataStore	يضيف JOIN
ds['col']	ColumnExpr	مرجع عمود
ds[['col1', 'col2']]	DataStore	تحديد الأعمدة

أمثلة:

# These do NOT trigger execution - they stay lazy
result = ds.filter(ds['age'] > 25)      # Returns DataStore
result = ds.select('name', 'age')        # Returns DataStore
result = ds['name']                      # Returns ColumnExpr
result = ds.groupby('city')              # Returns LazyGroupBy

---

​

التنفيذ على ثلاث مراحل

تتبع عمليات DataStore نموذج تنفيذ على ثلاث مراحل:

​

المرحلة 1: بناء استعلام SQL (مؤجَّل)

تتراكم العمليات التي يمكن التعبير عنها بلغة SQL:

result = (ds
    .filter(ds['status'] == 'active')   # WHERE
    .select('user_id', 'amount')         # SELECT
    .groupby('user_id')                  # GROUP BY
    .agg({'amount': 'sum'})              # SUM()
    .sort('sum', ascending=False)        # ORDER BY
    .limit(10)                           # LIMIT
)
# All compiled into one SQL query

​

المرحلة 2: نقطة التنفيذ

عند تحقّق المُشغِّل، يُنفَّذ استعلام SQL المتراكم:

# Execution triggered here
df = result.to_df()  
# The single optimized SQL query runs now

​

المرحلة 3: عمليات DataFrame (إن وُجدت)

إذا أتبعتَ التنفيذ بعمليات تقتصر على pandas فقط:

# Mixed operations
result = (ds
    .filter(ds['amount'] > 100)          # Phase 1: SQL
    .to_df()                             # Phase 2: Execute
    .pivot_table(...)                    # Phase 3: pandas
)

---

​

عرض خطط التنفيذ

استخدم explain() لمعرفة ما سيجري تنفيذه:

Query

ds = pd.read_csv("sales.csv")

query = (ds
    .filter(ds['amount'] > 1000)
    .groupby('region')
    .agg({'amount': ['sum', 'mean']})
)

# View execution plan
query.explain()

Response

Pipeline:
  1. Source: file('sales.csv', 'CSVWithNames')
  2. Filter: amount > 1000
  3. GroupBy: region
  4. Aggregate: sum(amount), avg(amount)

Generated SQL:
SELECT region, SUM(amount) AS sum, AVG(amount) AS mean
FROM file('sales.csv', 'CSVWithNames')
WHERE amount > 1000
GROUP BY region

استخدم verbose=True للحصول على مزيد من التفاصيل:

query.explain(verbose=True)

راجع تصحيح الأخطاء: explain() للاطلاع على الوثائق الكاملة.

---

​

التخزين المؤقت

يخزّن DataStore نتائج التنفيذ مؤقتًا لتجنّب تكرار الاستعلامات.

​

كيف يعمل التخزين المؤقت

from pathlib import Path
Path("data.csv").write_text("""\
name,age,city,salary,department
Alice,25,NYC,55000,Engineering
Bob,30,LA,65000,Product
Charlie,35,NYC,80000,Engineering
Diana,28,SF,70000,Design
Eve,42,NYC,95000,Product
""")

ds = pd.read_csv("data.csv")
result = ds.filter(ds['age'] > 25)

# First access - executes query
print(result.shape)  # Executes and caches

# Second access - uses cache
print(result.columns)  # Uses cached result

# Third access - uses cache
df = result.to_df()  # Uses cached result

​

إبطال صلاحية ذاكرة التخزين المؤقت

تُبطَل صلاحية ذاكرة التخزين المؤقت عندما تُجري العمليات تعديلات على DataStore:

result = ds.filter(ds['age'] > 25)
print(result.shape)  # Executes, caches

# New operation invalidates cache
result2 = result.filter(result['city'] == 'NYC')
print(result2.shape)  # Re-executes (different query)

​

التحكم اليدوي في ذاكرة التخزين المؤقت

# Clear cache
ds.clear_cache()

# Disable caching
from chdb.datastore.config import config
config.set_cache_enabled(False)

---

​

دمج عمليات SQL وPandas

يدير DataStore بذكاء العمليات التي تجمع بين SQL وPandas:

​

العمليات المتوافقة مع SQL

تُحوَّل هذه إلى SQL:

• filter(), where()
• select()
• groupby(), agg()
• sort(), orderby()
• limit(), offset()
• join(), union()
• distinct()
• عمليات على الأعمدة (العمليات الحسابية، والمقارنة، ودوال السلاسل النصية)

​

عمليات خاصة بـ Pandas فقط

تؤدي هذه العمليات إلى بدء التنفيذ وتستخدم pandas:

• apply() مع دوال مخصّصة
• pivot_table() مع عمليات تجميع معقّدة
• stack(), unstack()
• عمليات على كائنات DataFrame بعد تنفيذها

​

مسارات العمل الهجينة

# SQL phase
result = (ds
    .filter(ds['amount'] > 100)      # SQL
    .groupby('category')              # SQL
    .agg({'amount': 'sum'})           # SQL
)

# Execution + pandas phase
result = (result
    .to_df()                          # Execute SQL
    .pivot_table(...)                 # pandas operation
)

---

​

اختيار محرّك التنفيذ

يمكن لـ DataStore تنفيذ العمليات باستخدام محرّكات مختلفة:

​

الوضع التلقائي (الافتراضي)

from chdb.datastore.config import config

config.set_execution_engine('auto')  # Default
# Automatically selects best engine per operation

​

فرض استخدام محرك chDB

config.set_execution_engine('chdb')
# All operations use ClickHouse SQL

​

فرض استخدام محرك pandas

config.set_execution_engine('pandas')
# All operations use pandas

راجع التهيئة: محرك التنفيذ لمزيد من التفاصيل.

---

​

اعتبارات الأداء

​

جيد: طبّق التصفية مبكرًا

# Good: Filter in SQL, then aggregate
result = (ds
    .filter(ds['date'] >= '2024-01-01')  # Reduces data early
    .groupby('category')
    .agg({'amount': 'sum'})
)

​

سيئ: طبّق التصفية متأخرًا

# Bad: Aggregate all, then filter
result = (ds
    .groupby('category')
    .agg({'amount': 'sum'})
    .to_df()
    .query('sum > 1000')  # Pandas filter after aggregation
)

​

جيد: اختر الأعمدة مبكرًا

# Good: Select columns in SQL
result = (ds
    .select('user_id', 'amount', 'date')
    .filter(ds['date'] >= '2024-01-01')
    .groupby('user_id')
    .agg({'amount': 'sum'})
)

​

جيد: دع SQL يتولى العمل

# Good: Complex aggregation in SQL
result = (ds
    .groupby('category')
    .agg({
        'amount': ['sum', 'mean', 'count'],
        'quantity': 'sum'
    })
    .sort('sum', ascending=False)
    .limit(10)
)
# One SQL query does everything

# Bad: Multiple separate queries
sums = ds.groupby('category')['amount'].sum().to_df()
means = ds.groupby('category')['amount'].mean().to_df()
# Two queries instead of one

---

​

ملخص أفضل الممارسات

1. سلسِل العمليات قبل التنفيذ - ابنِ الاستعلام بالكامل، ثم شغِّله مرة واحدة
2. طبّق التصفية مبكرًا - قلّل البيانات عند المصدر
3. حدّد الأعمدة التي تحتاجها فقط - يحسّن تقليم الأعمدة الأداء
4. استخدم explain() لفهم آلية التنفيذ - صحّح الأخطاء قبل التشغيل
5. دع SQL يتولى التجميعات - ClickHouse مُحسَّن لهذا الغرض
6. انتبه إلى مشغّلات التنفيذ - تجنّب التشغيل المبكر غير المقصود
7. استخدم ذاكرة التخزين المؤقت بحكمة - افهم متى تُبطَل صلاحية ذاكرة التخزين المؤقت