Claude trong ETL Pipeline — Xử lý dữ liệu phi cấu trúc thành structured data

ETL (Extract, Transform, Load) là quy trình nền tảng trong data engineering. Bước Transform truyền thống hoạt động tốt với dữ liệu có cấu trúc — chuyển đổi format, mapping fields, aggregation. Nhưng khi nguồn dữ liệu là phi cấu trúc hoặc bán cấu trúc — invoices dạng PDF, emails, hợp đồng, báo cáo tài chính — các kỹ thuật transform thông thường trở nên bất lực. Claude API mở ra khả năng xử lý dữ liệu phi cấu trúc một cách có hệ thống, chính xác và có thể scale được.

Claude API trong bước Transform

Trong kiến trúc ETL truyền thống, bước Transform thực hiện các phép biến đổi deterministic: parse CSV, map columns, convert data types, apply business rules. Khi đưa Claude API vào bước Transform, bạn thêm khả năng xử lý ngôn ngữ tự nhiên — đọc hiểu văn bản, trích xuất thông tin từ format không cố định và chuyển thành structured output.

Kiến trúc pipeline

Pipeline tích hợp Claude API thường có kiến trúc sau:

• Extract: Thu thập raw data từ sources (S3, email server, file system, APIs). Không thay đổi so với ETL truyền thống
• Pre-process: Chuẩn bị data cho Claude — convert PDF sang text, extract email body, chunk documents lớn. Bước này giúp giảm token usage và tăng accuracy
• Transform (Claude API): Gửi pre-processed text cho Claude API với structured output schema. Claude trích xuất thông tin và trả về JSON theo schema định sẵn
• Post-process: Validate output từ Claude, apply business rules bổ sung, handle edge cases
• Load: Lưu structured data vào destination (database, data warehouse, API). Không thay đổi so với ETL truyền thống

Ví dụ thực tế: Trích xuất thông tin từ invoice

Đây là use case phổ biến nhất khi tích hợp Claude vào ETL pipeline. Hóa đơn từ nhiều nhà cung cấp có format khác nhau — không thể dùng regex hay template matching.

Bạn là một hệ thống trích xuất thông tin hóa đơn.
Từ nội dung hóa đơn bên dưới, trích xuất thông tin
theo JSON schema sau:

{
  "invoice_number": "string",
  "invoice_date": "YYYY-MM-DD",
  "due_date": "YYYY-MM-DD hoặc null",
  "vendor": {
    "name": "string",
    "tax_id": "string hoặc null",
    "address": "string hoặc null"
  },
  "buyer": {
    "name": "string",
    "tax_id": "string hoặc null",
    "address": "string hoặc null"
  },
  "line_items": [
    {
      "description": "string",
      "quantity": "number",
      "unit_price": "number",
      "total": "number",
      "vat_rate": "number hoặc null"
    }
  ],
  "subtotal": "number",
  "vat_amount": "number hoặc null",
  "total_amount": "number",
  "currency": "string (VND, USD, EUR)",
  "payment_info": {
    "bank_name": "string hoặc null",
    "account_number": "string hoặc null",
    "account_holder": "string hoặc null"
  },
  "confidence": "high | medium | low",
  "notes": "string - ghi chú nếu có thông tin không rõ ràng"
}

Quy tắc:
- Nếu không tìm thấy field, dùng null
- Số tiền luôn là number (không có dấu phẩy hoặc chấm ngàn)
- Ngày tháng luôn format YYYY-MM-DD
- confidence: high nếu tất cả thông tin rõ ràng,
  medium nếu phải suy luận 1-2 fields,
  low nếu nhiều thông tin thiếu hoặc mờ

Chỉ trả về JSON, không giải thích thêm.

---
NOI DUNG HOA DON:
[Nội dung invoice text/OCR output]

Xử lý batch invoices với Python

Trong thực tế, bạn cần xử lý hàng trăm hoặc hàng nghìn invoices. Dưới đây là cách tổ chức code Python để xử lý batch với rate limiting và error handling.

Viết Python script xử lý batch invoices với Claude API:

Requirements:
1. Đọc PDF invoices từ S3 bucket
2. Convert PDF sang text (dùng pdfplumber hoặc PyPDF2)
3. Gửi cho Claude API để extract structured data
4. Rate limiting: max 50 requests/minute (Tier 1)
5. Error handling:
   - Retry với exponential backoff cho API errors
   - Log failed extractions cho manual review
   - Continue processing khi 1 invoice fail
6. Output: Append results vào PostgreSQL table
7. Tracking: Log processing status, duration, cost per invoice
8. Resume capability: skip đã xử lý nếu re-run

Tech stack:
- Python 3.11
- anthropic SDK
- boto3 cho S3
- psycopg2 cho PostgreSQL
- pdfplumber cho PDF parsing

Cung cấp production-ready code với proper logging.

Claude sẽ sinh ra script hoàn chỉnh bao gồm connection pooling cho database, structured logging, progress tracking và cost estimation. Đặc biệt, Claude sẽ implement idempotency check để tránh xử lý lại invoice đã extract thành công.

Airflow + Claude Pipeline Architecture

Với pipeline quy mô lớn, Apache Airflow (hoặc tương đương như Dagster, Prefect) giúp orchestrate, schedule và monitor toàn bộ quy trình.

Thiết kế Airflow DAG cho ETL pipeline xử lý invoices:

DAG structure:
1. Task: check_new_files
   - Scan S3 bucket cho PDF mới (dùng S3 sensor hoặc list)
   - Output: danh sách file paths

2. Task: extract_text (dynamic task mapping)
   - Convert mỗi PDF sang text
   - Upload text file lên S3 (staging area)

3. Task: transform_with_claude (dynamic task mapping)
   - Gửi text cho Claude API
   - Parse JSON response
   - Validate output against schema
   - Rate limiting: 50 req/min across all parallel tasks

4. Task: validate_data
   - Cross-check extracted data (tổng = subtotal + VAT)
   - Flag invoices cần manual review
   - Business rules validation

5. Task: load_to_warehouse
   - Insert vào PostgreSQL/BigQuery
   - Update processing status

6. Task: notify
   - Slack notification: processed count, failed count, cost
   - Email report cho finance team

Schedule: Mỗi giờ
Concurrency: max 5 parallel Claude API calls
Retries: 3 lần cho API tasks, 1 lần cho khác

Hãy viết Airflow DAG code hoàn chỉnh (Airflow 2.7+,
TaskFlow API).

Email Processing Pipeline

Một use case phổ biến khác là trích xuất structured data từ emails — đặc biệt hữu ích cho customer support, order processing hoặc lead generation.

Thiết kế prompt cho bước Transform trong email processing pipeline.

Input: raw email (headers + body + attachments metadata)

Output schema:
{
  "email_type": "inquiry | complaint | order | feedback | spam | other",
  "priority": "urgent | high | normal | low",
  "sentiment": "positive | neutral | negative",
  "customer": {
    "name": "string hoặc null",
    "email": "string",
    "phone": "string hoặc null",
    "company": "string hoặc null"
  },
  "summary": "string - tóm tắt 1-2 câu",
  "action_items": ["string"],
  "entities": {
    "product_names": ["string"],
    "order_ids": ["string"],
    "dates_mentioned": ["YYYY-MM-DD"],
    "amounts": [{"value": "number", "currency": "string"}]
  },
  "suggested_department": "sales | support | billing | shipping | other",
  "language": "vi | en | other",
  "requires_human_review": "boolean",
  "review_reason": "string hoặc null"
}

Yêu cầu:
- Xử lý email tiếng Việt và tiếng Anh
- Detect forwarded emails và extract original sender
- Ignore email signatures và disclaimers
- Handle HTML emails (đã stripped thành text)
- Flag emails có nội dung nhạy cảm cần human review

Data Quality Validation

Output từ Claude API là probabilistic, không deterministic như traditional ETL. Vì vậy, data quality validation trở nên đặc biệt quan trọng.

Thiết kế data quality framework cho Claude API output
trong ETL pipeline:

Validation layers:

1. SCHEMA VALIDATION:
   - JSON schema compliance
   - Required fields present
   - Data types correct
   - Enum values valid

2. BUSINESS RULES:
   - Tổng line items = subtotal
   - subtotal + VAT = total
   - Date logic (invoice_date <= due_date)
   - Tax ID format validation (VN: 10 hoặc 13 digits)

3. CROSS-REFERENCE:
   - Vendor name match với vendor database (fuzzy match)
   - Duplicate invoice detection
   - Amount range check (flag if > 2 standard deviations)

4. CONFIDENCE-BASED ROUTING:
   - High confidence: auto-load vào warehouse
   - Medium confidence: auto-load + flag for spot check
   - Low confidence: queue for manual review

5. MONITORING:
   - Track extraction accuracy over time
   - Alert khi error rate > threshold
   - A/B test giữa prompt versions

Viết Python code cho validation framework này.

Cost Analysis và Optimization

Chi phí là yếu tố quan trọng khi dùng Claude API trong ETL pipeline, đặc biệt với batch processing quy mô lớn. Hiểu cách tính và tối ưu chi phí giúp bạn xây dựng pipeline bền vững.

Ước tính chi phí

Tính chi phí xử lý invoices bằng Claude API:

Giả định:
- 10,000 invoices/tháng
- Trung bình mỗi invoice: 800 tokens input (text sau OCR)
- System prompt: 500 tokens (schema + instructions)
- Output trung bình: 400 tokens (JSON response)

Model options:
1. Claude Sonnet: input $3/MTok, output $15/MTok
2. Claude Haiku: input $0.25/MTok, output $1.25/MTok

Hãy tính:
1. Chi phí mỗi invoice cho từng model
2. Chi phí tháng cho 10,000 invoices
3. So sánh accuracy vs cost giữa 2 models
4. Break-even point so với manual data entry
   (giả sử nhân viên lương 10 triệu VND/tháng,
    xử lý 200 invoices/ngày)
5. Strategies giảm chi phí:
   - Prompt caching
   - Batch API
   - Model routing (dùng Haiku cho simple invoices,
     Sonnet cho complex ones)

Prompt optimization cho cost

Giảm token count trong prompt và response trực tiếp giảm chi phí. Claude có thể giúp bạn tối ưu prompt mà không giảm accuracy.

Tối ưu prompt sau để giảm token count mà giữ nguyên accuracy:

[Dán prompt hiện tại]

Current stats:
- System prompt: 500 tokens
- Average input: 800 tokens
- Average output: 400 tokens
- Accuracy: 95%

Target:
- Giảm system prompt xuống dưới 300 tokens
- Giảm output tokens (compact JSON, shorter field names)
- Giữ accuracy >= 94%

Strategies cần xem xét:
1. Rút gọn instructions (loại ví dụ, giữ rules)
2. Shorter field names trong JSON schema
3. Output chỉ các fields có giá trị (skip nulls)
4. Enum mapping (1=high, 2=medium thay vì full string)
5. Response format: JSON Lines thay vì pretty-printed JSON

PDF Extraction Pipeline

PDF là format phổ biến nhất cho tài liệu kinh doanh. Xử lý PDF đòi hỏi bước pre-processing trước khi gửi cho Claude.

Thiết kế PDF extraction pipeline với các loại tài liệu:

1. Invoices (hóa đơn): structured, tabular data
2. Contracts (hợp đồng): long-form text, key clauses
3. Financial reports: tables + charts + narrative
4. Resumes/CVs: semi-structured, varied formats

Cho mỗi loại tài liệu, thiết kế:

A. Pre-processing strategy:
   - PDF to text tool (pdfplumber vs PyPDF2 vs OCR)
   - Table detection và extraction
   - Image handling (charts, signatures)
   - Multi-page document chunking

B. Claude prompt template:
   - Extraction schema (JSON)
   - Instructions specific cho document type
   - Few-shot examples (nếu cần)

C. Post-processing:
   - Validation rules
   - Confidence scoring
   - Error handling

D. Estimated accuracy và cost per document

Ưu tiên invoice pipeline vì dùng nhiều nhất.
Database: PostgreSQL, Cloud: AWS

Error Handling và Recovery

Pipeline production cần xử lý nhiều loại lỗi khác nhau. Claude có thể giúp bạn thiết kế error handling strategy toàn diện.

Thiết kế error handling cho ETL pipeline dùng Claude API:

Error categories:

1. API ERRORS:
   - Rate limit (429): retry với backoff
   - Server error (500, 503): retry với backoff
   - Authentication error (401): alert, stop pipeline
   - Context length exceeded (400): chunk document, retry

2. EXTRACTION ERRORS:
   - Invalid JSON response: retry với stricter prompt
   - Missing required fields: retry hoặc flag for review
   - Low confidence: route to manual review queue

3. DATA ERRORS:
   - Duplicate records: skip hoặc update
   - Validation failures: log và continue
   - Foreign key violations: queue for later processing

4. INFRASTRUCTURE ERRORS:
   - S3 access errors: retry
   - Database connection errors: connection pool retry
   - Memory errors (large PDF): skip, log, alert

Cho mỗi category:
- Retry strategy (max retries, backoff formula)
- Fallback behavior
- Logging format
- Alert conditions
- Dead letter queue design

Viết Python error handling framework.

Monitoring và Observability

Pipeline ETL dùng Claude API cần monitoring chặt chẽ hơn pipeline thông thường vì output là non-deterministic.

Thiết kế monitoring dashboard cho Claude ETL pipeline:

Metrics cần track:

1. THROUGHPUT:
   - Documents processed / hour
   - Documents in queue (backlog)
   - Processing latency (end-to-end, per step)

2. QUALITY:
   - Extraction accuracy (validated sample)
   - Confidence score distribution
   - Manual review rate
   - Error rate by document type

3. COST:
   - API cost per document (input + output tokens)
   - Total daily/monthly cost
   - Cost trend (increasing/decreasing)
   - Cost per document type

4. API HEALTH:
   - Request latency (P50, P95, P99)
   - Error rate by type (429, 500, etc.)
   - Token usage vs limits
   - Cache hit rate (if using prompt caching)

5. ALERTS:
   - Error rate > 5% trong 1 giờ
   - Processing backlog > 1000 documents
   - API cost > daily budget
   - Accuracy drop > 3% so với baseline

Stack: Prometheus + Grafana hoặc Datadog
Hãy đề xuất metric names, labels và Grafana dashboard layout.

Prompt Templates tổng hợp

Generic document extraction

Extract structured data từ document sau.

Document type: [invoice/contract/report/email/other]
Language: [vi/en]

Output schema:
[Dán JSON schema]

Rules:
- Return valid JSON only
- Use null for missing fields
- Dates in YYYY-MM-DD format
- Numbers without formatting (no commas)
- Include confidence field (high/medium/low)

Document content:
[Document text]

Pipeline design consultation

Tôi cần xây dựng ETL pipeline để xử lý [loại data]:

Sources: [liệt kê data sources]
Volume: [số documents/ngày]
Output: [destination database/warehouse]
Latency requirement: [real-time/hourly/daily]
Budget: [monthly budget cho API costs]

Hãy thiết kế:
1. Pipeline architecture diagram (text)
2. Technology stack recommendation
3. Claude API integration points
4. Cost estimation
5. Scaling strategy
6. Monitoring plan

Scaling Patterns

Khi volume dữ liệu tăng từ hàng trăm lên hàng chục nghìn documents mỗi ngày, pipeline cần được thiết kế để scale hiệu quả. Claude có thể giúp bạn chọn scaling pattern phù hợp.

Pipeline hiện tại xử lý 500 documents/ngày, cần scale lên 10,000/ngày.

Bottlenecks hiện tại:
- Claude API rate limit: 50 requests/minute
- PDF text extraction: 5 seconds/document
- Database writes: không phải bottleneck

Hãy đề xuất scaling strategy:

1. Horizontal scaling:
   - Parallel workers (Celery, AWS Lambda, K8s Jobs)
   - Partition documents theo type/priority
   - Queue-based architecture (SQS, RabbitMQ)

2. API optimization:
   - Batch API (50% cheaper, higher throughput)
   - Prompt caching cho system prompt
   - Model routing: Haiku cho simple, Sonnet cho complex
   - Request batching: group multiple short documents in 1 call

3. Pre-processing optimization:
   - Parallel PDF extraction
   - Skip extraction nếu PDF đã có text layer
   - Cache OCR results cho duplicate documents

4. Architecture:
   - Synchronous vs asynchronous processing
   - Priority queues cho urgent documents
   - Dead letter queue cho failures
   - Backpressure handling

Estimate throughput và cost cho mỗi approach.

Best Practices khi dùng Claude API trong ETL

• Structured output: Luôn định nghĩa JSON schema rõ ràng trong prompt. Sử dụng tool use hoặc JSON mode nếu API hỗ trợ để đảm bảo output format chính xác
• Idempotency: Thiết kế pipeline để có thể re-run an toàn. Track processed documents bằng unique ID và skip nếu đã xử lý
• Model selection: Dùng Haiku cho documents đơn giản (invoices chuẩn), Sonnet cho documents phức tạp (contracts, reports). Intelligent routing giảm chi phí đáng kể
• Prompt versioning: Version control prompts giống như code. Mỗi thay đổi prompt cần test trên sample data trước khi deploy
• Human-in-the-loop: Luôn có quy trình manual review cho low-confidence extractions. Feedback từ manual review dùng để cải thiện prompt
• Batch API: Nếu không cần real-time, dùng Batch API để giảm 50% chi phí và tránh rate limit issues

Bước tiếp theo

Bạn đã nắm được cách tích hợp Claude API vào ETL pipeline để xử lý dữ liệu phi cấu trúc — từ thiết kế kiến trúc, viết extraction prompts, xử lý batch, đến monitoring và cost optimization. Bước tiếp theo là xây dựng proof-of-concept với một loại document cụ thể (khuyến nghị bắt đầu với invoices) và đo lường accuracy trước khi scale. Khám phá thêm các hướng dẫn tại Thư viện Ứng dụng Claude.