ایجاد خطوط لوله داده مقیاس پذیر با پایتون – یک راهنمای کامل.

خطوط لوله داده چیست

یک خط لوله داده به یک سری مراحل استفاده شده برای خودکارسازی مهاجرت داده ها از یک منبع به مقصد خود اشاره دارد. بعضی اوقات ، تحول در کنار مهاجرت انجام می شود تا اطمینان حاصل شود که داده ها ساختار یافته و برای تجزیه و تحلیل تمیز هستند.

اجزای خط لوله

اجزای یک خط لوله به عناصری که برای تشکیل یک خط لوله داده جمع می شوند ، اشاره دارد. اینها شامل:

1. منابع داده – اینها می توانند شامل بانکهای اطلاعاتی ، پرونده های CSV ، API و سایر قالب های پرونده باشند.
2. روشهای مصرف داده – اینها به نحوه بارگیری داده ها در خط لوله اشاره دارند. دو روش اصلی وجود دارد:
   • پردازش دسته
   • پردازش جریان
3. داده های پردازش – این به تکنیک ها و ابزارهای مورد استفاده برای تبدیل داده ها اشاره دارد.
4. ذخیره داده ها – این به محل ذخیره داده ها ، از جمله انبارهای داده ، دریاچه های داده و غیره اشاره دارد. این معمولاً مقصد نهایی داده ها است.

عملکردهای کلیدی یک خط لوله

توابع اصلی یک خط لوله شامل موارد زیر است:

1. استخراج کردن
2. تغییر شکل
3. بار

ملاحظات هنگام طراحی خط لوله

هنگام طراحی خط لوله داده ، عوامل زیر را باید در نظر گرفت:

1. مقیاس پذیری
2. حفظ قابلیت
3. امنیت
4. اتهام اتوماسیون

اجرای Python ETL

در این بخش خواهیم دید که چگونه یک خط لوله ETL ساده را برای خواندن داده ها از یک فایل CSV و یک API اجرا کنیم ، سپس داده ها را در یک پایگاه داده PostgreSQL بنویسید.

1. خواندن از یک پرونده CSV

قبل از ساختن خط لوله ETL برای خواندن از یک فایل CSV ، ما نیاز داریم:

• یک فایل CSV (با استفاده از Mockaroo برای داده های ساختگی تولید شده است).
• یک پایگاه داده PostgreSQL (با استفاده از AIVEN و با استفاده از DBEAVER ایجاد شده است).

اسکریپت پایتون زیر نحوه خواندن از یک فایل CSV را نشان می دهد و داده ها را در یک پایگاه داده PostgreSQL ذخیره می کند:

import pandas as pd
from sqlalchemy import create_engine

# Create a connection to PostgreSQL
engine = create_engine("postgresql://username:password@localhost:5432/etl_db")

# Read CSV file
df = pd.read_csv("sales.csv")

# Data Cleaning and Transformation
df = df.dropna()
df = df.rename(columns={'id': 'sales_id'})

# Load data into PostgreSQL
df.to_sql("sales", engine, if_exists="append", index=False)

کد فوق با موفقیت داده ها را از یک فایل CSV به پایگاه داده منتقل می کند و نشان می دهد که چگونه یک خط لوله ETL کار می کند.

2. خواندن از API

ما داده ها را از این API دریافت خواهیم کرد: نمونه داده های JSON. این داده های ساختگی نمایندگان کارمندان یک شرکت داستانی را نشان می دهد.

فرآیند ETL به شرح زیر است:

import requests
import pandas as pd
from sqlalchemy import create_engine

# API URL
url = "https://raw.githubusercontent.com/LuxDevHQ/LuxDevHQDataEngineeringGuide/refs/heads/main/samplejson.json"

# Fetch data from API
response = requests.get(url)
data = response.json()

# Transform data
df = pd.DataFrame(data)
df = df[['name', 'position', 'country']]
df.columns = ['full_name', 'role', 'country']

# Load data into PostgreSQL
engine = create_engine("postgresql://username:password@localhost:5432/etl_db")
df.to_sql("staff_data", engine, if_exists="append", index=False)

پایان

ما در مورد خطوط لوله داده ، مؤلفه های آنها ، کارکردهای کلیدی ، ملاحظات طراحی و نحوه اجرای یک خط لوله ETL ساده برای خواندن داده ها از یک فایل CSV و یک API در یک پایگاه داده PostgreSQL آموخته ایم.