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

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

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

توابع کلیدی یک خط لوله داده

• استخراج کردن: فرآیند بازیابی داده های خام از چندین منبع (پایگاه داده ، API ، پرونده ها و غیره).
• تغییر شکل: تمیز ، فیلتر ، جمع آوری یا تغییر داده ها برای مفید شدن آن.
• بار: داده های تبدیل شده را در یک سیستم هدف ، مقصد (انبار داده ، پایگاه داده یا ابزار تحلیلی) ذخیره کنید.

اجرای Python ETL

در این بخش نمونه های خط لوله ETL را با استفاده از پایتون پیاده سازی خواهیم کرد:

خواندن از CSV و نوشتن به postgresql

1. داده های فروش را تولید کنید (من از Mockaroo برای تولید داده های ساختگی استفاده کرده ام).
2. یک پایگاه داده postgresql ایجاد کنیدبشر من استفاده کرده ام aiven.io برای ذخیره داده های تبدیل شده من و DBeaver برای اتصال به پایگاه داده.
3. فرآیند ETL را با استفاده از پایتون پیاده سازی کنیدبشر حتماً کتابخانه های لازم را نصب کنید (pandas وت sqlalchemy).

import pandas as pd
from sqlalchemy import create_engine

# Database connection
engine = create_engine("postgresql://username:password@localhost:5432/etl_db")

# Extract
df = pd.read_csv("sales.csv")

# Transform the data
df = df.rename(columns={'id': 'sales_id'})  # Rename columns
df = df.dropna()  # Drop null values

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

print("Data successfully loaded into PostgreSQL!")

خواندن از API و نوشتن به پایگاه داده

ما قصد داریم از این نقطه پایانی برای واکشی داده های JSON (داده های کارکنان یک شرکت ساختگی) استفاده کنیم: نمونه داده های JSON.

کد پایتون برای واکشی و بارگیری داده های API:

import requests
import json
import pandas as pd
from sqlalchemy import create_engine

# Step 1: Fetch data from API
url = "https://raw.githubusercontent.com/LuxDevHQ/LuxDevHQDataEngineeringGuide/refs/heads/main/samplejson.json"
response = requests.get(url)
data = response.json()

# Step 2: Extract - Convert JSON data to DataFrame
df = pd.DataFrame(data)

# Step 3: Transform - Clean and reshape the data
df = df[['name', 'position', 'country']]  # Select relevant columns
df.columns = ['full_name', 'role', 'country']  # Rename columns for clarity

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

print("API Data Loaded Successfully!")

تبریک می گویم! 🎉

اگر هر مرحله را به درستی دنبال کرده اید ، با استفاده از پایتون یک خط لوله داده ETL را با موفقیت اجرا کرده اید!

این مثال مفید نشان می دهد که چگونه می توان فرایند انتقال داده ها را از پرونده های CSV و API ها به یک پایگاه داده ، به صورت خودکار انجام داد و گردش کار پردازش داده های شما را ساده تر و کارآمدتر و مقیاس پذیر تر کرد.