Trong bài viết này, tôi sẽ chia sẻ cách tôi xây dựng một CSV ETL pipeline hoàn chỉnh sử dụng Python kết hợp với AI để tự động hóa quy trình làm sạch, chuyển đổi và nhập dữ liệu. Đây là giải pháp tôi đã triển khai thực tế cho dự án Tardis với khối lượng xử lý hơn 500,000 records mỗi ngày.
Tổng quan kiến trúc ETL Pipeline
Pipeline của chúng ta sẽ bao gồm 3 stage chính:
- Extract (E): Đọc và phân tích file CSV từ nhiều nguồn
- Transform (T): Làm sạch, chuẩn hóa và chuyển đổi dữ liệu bằng AI
- Load (L): Đẩy dữ liệu đã xử lý vào database đích
Cài đặt môi trường và dependencies
Trước tiên, hãy cài đặt các thư viện cần thiết:
pip install pandas numpy python-dotenv requests sqlalchemy
pip install sqlalchemy psycopg2-binary # PostgreSQL
pip install sqlalchemy pymysql # MySQL
pip install pandas-openpyxl xlrd # Excel supportModule 1: Kết nối HolySheep AI cho Data Cleaning thông minh
Điểm mấu chốt của pipeline này là sử dụng AI để tự động phát hiện và sửa lỗi dữ liệu. Tôi sử dụng HolySheep AI vì:
- Độ trễ trung bình chỉ <50ms — nhanh hơn 10 lần so với các provider khác
- Tỷ giá ¥1=$1 — tiết kiệm 85%+ chi phí
- Hỗ trợ thanh toán WeChat/Alipay cho người dùng Việt Nam
- Tín dụng miễn phí khi đăng ký — không rủi ro để thử nghiệm
import os
import requests
import json
from typing import Dict, List, Any
class HolySheepAIClient:
"""Client cho HolySheep AI API - tối ưu cho ETL data cleaning"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def clean_data_with_ai(self, data_batch: List[Dict],
schema: Dict) -> List[Dict]:
"""
Sử dụng AI để làm sạch và chuẩn hóa dữ liệu
Args:
data_batch: Danh sách records cần làm sạch
schema: Schema mô tả cấu trúc dữ liệu mong đợi
Returns:
List[Dict]: Dữ liệu đã được làm sạch
"""
prompt = f"""Bạn là chuyên gia data cleaning. Hãy làm sạch dữ liệu sau:
Schema mong đợi: {json.dumps(schema, ensure_ascii=False)}
Dữ liệu đầu vào:
{json.dumps(data_batch, ensure_ascii=False, indent=2)}
Yêu cầu:
1. Sửa lỗi chính tả, format không nhất quán
2. Chuẩn hóa các trường theo schema
3. Xử lý giá trị NULL, empty strings
4. Đảm bảo kiểu dữ liệu đúng (string, number, date)
5. Trả về JSON array chỉ với dữ liệu đã làm sạch
Output format: JSON array"""
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 4000
},
timeout=30
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
result = response.json()
cleaned_data = json.loads(result['choices'][0]['message']['content'])
return cleaned_data
def detect_schema_from_sample(self, sample_data: List[Dict]) -> Dict:
"""Tự động phát hiện schema từ sample data"""
prompt = f"""Phân tích dữ liệu mẫu sau và trả về JSON schema mô tả cấu trúc:
{json.dumps(sample_data, ensure_ascii=False, indent=2)}
Trả về JSON với format:
{{
"fields": {{
"field_name": {{
"type": "string|number|date|boolean",
"required": true|false,
"description": "mô tả"
}}
}}
}}"""
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1
}
)
return json.loads(response.json()['choices'][0]['message']['content'])Module 2: Data Extractor cho CSV files
import pandas as pd
from pathlib import Path
from typing import Iterator, Dict, Any
import chardet
class CSVExtractor:
"""Trình đọc CSV với tự động