譯者 | 布加迪
審校 | 重樓
高質量的數據在數據科學中非常重要，但這類數據常常來自許多地方，格式混亂。一些數據來自數據庫，另一些數據來自文件或網站。這些原始數據很難立即使用，因此我們需要先對其進行清理和組織。
ETL是幫助完成這項任務的過程。ETL代表提取、轉換和加載。提取意味著從不同的來源收集數據。轉換意味著清理和格式化數據。加載意味著將數據存儲在數據庫中以便訪問。構建ETL管道使這個過程實現了自動化。穩健的ETL管道可以節省時間，并確保數據可靠。
我們在本文中將了解如何為數據科學項目構建ETL管道。
什么是ETL管道？
ETL管道將數據從源端移動到目標端。它分為三個階段：
1. 提取：從多個來源（比如數據庫或文件）收集數據。
2. 轉換：清理和轉換數據以供分析。
3. 加載：將清理后的數據存儲在數據庫或其他系統中。
為什么ETL管道很重要？
ETL管道很重要，有這幾個原因：
- 數據質量：轉換通過處理缺失值和修復錯誤來幫助清理數據。
- 數據可訪問性：ETL管道將來自許多來源的數據放到一個地方，以便訪問。
- 自動化：管道自動執行重復性任務，讓數據科學家可以專注于分析。
現在，讓我們用Python構建一個簡單的ETL管道。
數據攝取
首先，我們需要得到數據。我們將從一個CSV文件中提取數據。
import pandas as pd
# Function to extract data from a CSV file
def extract_data(file_path):
try:
data = pd.read_csv(file_path)
print(f"Data extracted from {file_path}")
return data
except Exception as e:
print(f"Error in extraction: {e}")
return None
# Extract employee data
employee_data = extract_data('/content/employees_data.csv')
# Print the first few rows of the data
if employee_data is not None:
print(employee_data.head())
數據轉換
收集數據后，我們需要對其進行轉換。這意味著要清理數據，并確保其正確。我們還將數據更改為可用于分析的格式。下面是一些常見的轉換：
- 處理缺失的數據：刪除或填寫缺失的值。
- 創建衍生的特征：創建新的列，比如工資區間或年齡組。
- 編碼類別：將部門名稱等數據更改為計算機可以使用的格式。
# Function to transform employee data
def transform_data(data):
try:
# Ensure salary and age are numeric and handle any errors
data['Salary'] = pd.to_numeric(data['Salary'], errors='coerce')
data['Age'] = pd.to_numeric(data['Age'], errors='coerce')
# Remove rows with missing values
data = data.dropna(subset=['Salary', 'Age', 'Department'])
# Create salary bands
data['Salary_band'] = pd.cut(data['Salary'], bins=[0, 60000, 90000, 120000, 1500000], labels=['Low', 'Medium', 'High', 'Very High'])
# Create age groups
data['Age_group'] = pd.cut(data['Age'], bins=[0, 30, 40, 50, 60], labels=['Young', 'Middle-aged', 'Senior', 'Older'])
# Convert department to categorical
data['Department'] = data['Department'].astype('category')
print("Data transformation complete")
return data
except Exception as e:
print(f"Error in transformation: {e}")
return None
employee_data = extract_employee_data('/content/employees_data.csv')
# Transform the employee data
if employee_data is not None:
transformed_employee_data = transform_data(employee_data)
# Print the first few rows of the transformed data
print(transformed_employee_data.head())
數據存儲
最后一步是將數據加載到數據庫中，使得用戶易于搜索和分析。
在本文中，我們使用SQLite。它是一種存儲數據的輕量級數據庫。我們將在SQLite數據庫中創建一個名為employees（員工）的表。然后，我們將轉換后的數據插入到該表中。
import sqlite3
# Function to load transformed data into SQLite database
def load_data_to_db(data, db_name='employee_data.db'):
try:
# Connect to SQLite database (or create it if it doesn't exist)
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
# Create table if it doesn't exist
cursor.execute('''
CREATE TABLE IF NOT EXISTS employees (
employee_id INTEGER PRIMARY KEY,
first_name TEXT,
last_name TEXT,
salary REAL,
age INTEGER,
department TEXT,
salary_band TEXT,
age_group TEXT
)
''')
# Insert data into the employees table
data.to_sql('employees', conn, if_exists='replace', index=False)
# Commit and close the connection
conn.commit()
print(f"Data loaded into {db_name} successfully")
# Query the data to verify it was loaded
query = "SELECT * FROM employees"
result = pd.read_sql(query, conn)
print("\nData loaded into the database:")
print(result.head()) # Print the first few rows of the data from the database
conn.close()
except Exception as e:
print(f"Error in loading data: {e}")
load_data_to_db(transformed_employee_data)
運行完整的ETL管道
現在我們已完成了提取、轉換和加載等步驟，就可以將它們組合起來了。這將創建一個完整的ETL管道。該管道將獲取員工數據，并清理和更改數據。最后，它將把數據保存在數據庫中。
def run_etl_pipeline(file_path, db_name='employee_data.db'):
# Extract
data = extract_employee_data(file_path)
if data is not None:
# Transform
transformed_data = transform_employee_data(data)
if transformed_data is not None:
# Load
load_data_to_db(transformed_data, db_name)
# Run the ETL pipeline
run_etl_pipeline('/content/employees_data.csv', 'employee_data.db')現在大功告成了：我們的ETL管道已實現完畢，現在就可以執行了。
ETL管道的幾個最佳實踐
下面是需要遵循的幾個最佳實踐，以便構建高效可靠的ETL管道：
1. 利用模塊化：將管道分解為更小的、可重用的函數。
2. 錯誤處理：在提取、轉換或加載期間為日志問題添加錯誤處理機制。
3. 優化性能：為大型數據集優化查詢和管理內存。
4. 自動化測試：自動測試轉換和數據格式，以確保準確性。
結語
ETL管道是任何數據科學項目的關鍵。它們有助于處理和存儲數據，以供進行準確的分析。我們演示了如何從CSV文件中獲取數據，然后我們清理和更改了數據，最后我們將其保存在SQLite數據庫中。
一個好的ETL管道可以確保數據井然有序。這種管道可加以改進，以處理更復雜的數據和存儲需求。它有助于創建可擴展且可靠的數據解決方案。
原文標題：Developing Robust ETL Pipelines for Data Science Projects，作者：Jayita Gulati
