etl-concurrent-framework 0.1.1

可擴展的ETL併發處理框架

ETL併發處理框架

高效能、可擴展的ETL（Extract-Transform-Load）併發處理框架，專為處理會計底稿和數據分析而設計。利用多線程和多進程技術，大幅提升數據處理效能，同時保持代碼清晰和可維護性。

✨ 特色功能

• 高效併發處理：針對I/O和CPU密集型任務分別採用多線程和多進程，實現最佳性能
• 自適應資源優化：基於數據大小和系統資源動態調整並發參數
• 增強錯誤恢復：完整的錯誤追蹤與異常處理，確保ETL流程穩定性
• 線程與進程安全：細粒度鎖機制和線程本地存儲保證併發安全
• 高級日誌系統：非阻塞式日誌處理和自動堆疊追蹤功能
• 靈活的處理策略：可切換的轉換邏輯，支持不同業務場景
• 多格式輸出支持：內建支持CSV、Excel、Parquet等多種數據格式
• 全面的性能監控：自動收集資源使用與處理性能統計

🛠️ 安裝與依賴

環境要求

• Python 3.8+
• pandas
• numpy
• openpyxl (用於Excel文件處理)
• pyarrow (用於Parquet文件處理)
• psutil (用於系統資源監控)

安裝

1. 複製儲存庫

git clone https://github.com/yourusername/etl-concurrent-framework.git
cd etl-concurrent-framework

2. 安裝依賴套件

pip install -r requirements.txt

📋 快速開始

基本用法

from core import ETLContext, ProcessingMode
from processors.extract import ExtractProcessor
from processors.transform import TransformProcessor
from processors.load import LoadProcessor
from orchestration.orchestrator import ETLOrchestratorWithOutput
from utils.resource_manager import ResourceManager

# 初始化資源管理器
resource_manager = ResourceManager()
resource_manager.start_monitoring()

# 初始化上下文與處理器
context = ETLContext()
extractor = ExtractProcessor(context) 
transformer = TransformProcessor(
    context=context,
    resource_manager=resource_manager
)
loader = LoadProcessor(context)

# 創建協調器
orchestrator = ETLOrchestratorWithOutput(
    context=context,
    extractor=extractor,
    transformer=transformer,
    loader=loader
)

# 執行ETL流程
success = orchestrator.run(
    data_dir='data/raw',  # 數據目錄
    file_pattern='vouchers_*.xlsx',  # 文件匹配模式
    processing_mode=ProcessingMode.CONCURRENT,  # 並行處理模式
    extract_params={'max_workers': 4},  # 提取參數
    transform_params={'max_workers': 4},  # 轉換參數
    load_params={'max_workers': 3},  # 載入參數
    reports=[  # 報表配置
        {
            'dimension': 'account_balance',
            'filename': 'data/reports/balance_sheet.xlsx',
            'params': {
                'groupby_cols': ['company', 'year', 'month', 'account_code'],
                'agg_dict': {
                    'debit_amount': 'sum',
                    'credit_amount': 'sum'
                },
                'post_process': lambda df: df.assign(
                    balance=df['debit_amount'] - df['credit_amount']
                )
            }
        }
    ],
    enable_auto_optimization=True  # 啟用自動優化
)

# 清理資源
resource_manager.stop_monitoring()

併發模式選項

# 串行處理模式（適合小數據量或調試）
orchestrator.run(
    processing_mode=ProcessingMode.SEQUENTIAL,
    # 其他參數...
)

# 並行處理模式（適合大數據量生產環境）
orchestrator.run(
    processing_mode=ProcessingMode.CONCURRENT,
    # 其他參數...
)

🏗️ 框架架構

核心組件

• ETLContext: 管理共享資源和狀態的上下文
• ExtractProcessor: 負責從各種來源讀取數據（使用多線程）
• TransformProcessor: 負責數據轉換和處理（使用多進程）
• LoadProcessor: 負責聚合和報表生成（使用多線程）
• OutputProcessor: 負責數據輸出（無聚合邏輯，使用多線程）
• ETLOrchestrator: 協調整個ETL流程的執行
• ResourceManager: 監控系統資源和優化處理參數

處理策略

框架使用策略模式實現可切換的數據轉換邏輯:

• DefaultSalesTransformStrategy: 銷售數據轉換邏輯
• AccountingTransformStrategy: 會計數據轉換邏輯
• 可自行擴展實現 TransformStrategy 接口

# 自定義轉換策略示例
class MyCustomStrategy(TransformStrategy):
    def transform(self, df, **kwargs):
        # 自定義轉換邏輯
        return transformed_df

# 使用自定義策略
transformer = TransformProcessor(
    context=context,
    strategy_class=MyCustomStrategy
)

🧠 並發安全與錯誤處理

併發安全機制

框架採用多層次的併發安全機制:

# 文件鎖 - 防止文件競爭
with file_lock_manager.get_lock(filename):
    df.to_excel(filename, **params)

# 統計信息更新鎖 - 保護共享狀態
with self._stats_lock:
    self.context.stats.file_processed(file_path, rows)

# 線程本地存儲 - 隔離線程數據
_thread_local.current_task = {'config': output_config}

錯誤恢復機制

完整的錯誤捕獲與記錄:

try:
    # 處理代碼
except Exception as e:
    logger.error(f"處理失敗: {str(e)}")
    logger.error(f"堆疊追蹤:\n{traceback.format_exc()}")
    # 更新錯誤統計
    self.context.stats.record_error(type(e).__name__)

🔧 進階用法

多格式輸出配置

# 定義多種輸出格式
output_configs = [
    # CSV輸出
    {
        'filename': 'data/processed/accounts.csv',
        'params': {
            'encoding': 'utf-8-sig',
            'index': False
        }
    },
    # Excel輸出
    {
        'filename': 'data/processed/accounts.xlsx',
        'params': {
            'sheet_name': '科目表',
            'freeze_panes': (1, 0)
        }
    },
    # Parquet輸出
    {
        'filename': 'data/processed/accounts.parquet',
        'params': {
            'compression': 'snappy'
        }
    }
]

# 在ETL流程中使用
orchestrator.run(
    # 其他參數...
    output_configs=output_configs,
    output_params={'max_workers': 3}
)

自定義報表設計

# 定義複雜報表
reports = [
    # 科目餘額表
    {
        'dimension': 'account_balance',
        'filename': 'data/reports/account_balance.xlsx',
        'params': {
            'groupby_cols': ['company', 'year', 'month', 'account_code', 'account_name'],
            'agg_dict': {
                'debit_amount': 'sum',
                'credit_amount': 'sum'
            },
            'post_process': lambda df: df.assign(
                balance=df['debit_amount'] - df['credit_amount']
            ),
            'write_params': {
                'sheet_name': '科目餘額表',
                'freeze_panes': (1, 0)
            }
        }
    }
]

選擇性處理階段

# 跳過載入階段，只執行提取、轉換和輸出
orchestrator.run(
    # 其他參數...
    skip_load=True,
    output_configs=output_configs
)

# 跳過轉換階段，直接從提取到輸出
orchestrator.run(
    # 其他參數...
    skip_transform=True,
    skip_load=True,
    output_configs=output_configs
)

性能比較

from orchestration.performance import PerformanceComparator

# 創建性能比較器
comparator = PerformanceComparator(orchestrator)

# 比較串行和並行模式的性能
comparator.compare(
    extract_params={'max_workers': 4},
    transform_params={'max_workers': 4},
    load_params={'max_workers': 3}
)

📊 會計數據處理

會計底稿處理示例

from accounting_logics import accounting_transform

# 設置會計轉換參數
transform_params = {
    'num_partitions': 4,
    'max_workers': 4,
    'custom_transform': accounting_transform
}

# 定義會計報表
accounting_reports = [
    # 傳票彙總表
    {
        'dimension': 'voucher_summary',
        'filename': 'data/accounting/reports/voucher_summary.xlsx',
        'params': {
            'groupby_cols': ['company', 'year', 'month', 'type'],
            'agg_dict': {
                'voucher_id': 'count',
                'amount': 'sum'
            },
            'post_process': lambda df: df.rename(
                columns={'voucher_id': 'voucher_count'}
            )
        }
    },
    # 科目餘額表
    {
        'dimension': 'account_balance',
        'filename': 'data/accounting/reports/account_balance.xlsx',
        'params': {
            'groupby_cols': ['company', 'year', 'month', 'account_code', 'account_name'],
            'agg_dict': {
                'debit_amount': 'sum',
                'credit_amount': 'sum'
            },
            'post_process': lambda df: df.assign(
                balance=df['debit_amount'] - df['credit_amount']
            )
        }
    }
]

# 使用會計轉換策略執行流程
transformer = TransformProcessor(
    context=context,
    strategy_class=AccountingTransformStrategy
)

orchestrator = ETLOrchestratorWithOutput(
    context=context,
    extractor=extractor,
    transformer=transformer,
    loader=loader
)

orchestrator.run(
    data_dir='data/accounting/raw',
    file_pattern='vouchers_*.xlsx',
    processing_mode=ProcessingMode.CONCURRENT,
    transform_params=transform_params,
    reports=accounting_reports
)

📈 實用範例情境

框架支持多種使用情境：

情境1: 單純讀取文件

from core import ETLContext
from processors.extract import ExtractProcessor

context = ETLContext()
extractor = ExtractProcessor(context)

# 並行讀取文件
combined_data = extractor.process_concurrent(
    file_paths=['data/file1.xlsx', 'data/file2.xlsx'],
    max_workers=4,
    parse_dates=['date']
)

情境2: 讀取與轉換

from core import ETLContext, ProcessingMode
from processors.extract import ExtractProcessor
from processors.transform import TransformProcessor
from orchestration.orchestrator import ETLOrchestratorWithOutput

context = ETLContext()
extractor = ExtractProcessor(context)
transformer = TransformProcessor(context)

orchestrator = ETLOrchestratorWithOutput(
    context=context,
    extractor=extractor,
    transformer=transformer
)

orchestrator.run(
    data_dir='data/raw',
    file_pattern='*.csv',
    skip_load=True,
    skip_output=True,
    transform_params={
        'num_partitions': 4,
        'max_workers': 4,
        'save_transformed': True,
        'transformed_path': 'data/transformed.csv'
    }
)

情境3: 完整ETL處理

完整的ETL流程見快速開始部分和會計數據處理部分。

🔍 性能優化建議

1. I/O密集型處理（提取和輸出階段）:

   • 增加 max_workers 參數值以提高並行度
   • 範例: extract_params={'max_workers': 8}
   • 使用多線程而非多進程，因為I/O受限操作不受GIL影響

2. CPU密集型處理（轉換階段）:

   • 設置合適的分區數和工作進程數，通常不超過CPU核心數
   • 範例: transform_params={'num_partitions': 4, 'max_workers': 4}
   • 減少分區數量可以降低合併開銷

3. 自動優化:

   • 啟用 enable_auto_optimization=True 參數
   • 框架會基於系統資源和數據大小動態調整參數
   • 動態監控CPU和記憶體使用率

4. 大型數據集:

   • 增加分區數以減少每個工作進程的記憶體使用
   • 範例: transform_params={'num_partitions': 8}
   • 考慮使用Parquet輸出以減少磁盤使用和提高後續讀取速度

🔒 併發安全性設計

框架設計上考慮多層次的併發安全性:

1. 文件級鎖定: 使用FileLockManager提供細粒度文件鎖，防止併發寫入衝突
2. 共享資源保護: 所有共享狀態操作都使用適當的鎖保護
3. 線程隔離: 使用threading.local()實現線程特定數據隔離
4. 計數器安全: 使用專用鎖保護進度和統計計數器
5. 跨進程安全: 針對多進程場景設計的資源分配和初始化機制

💾 日誌系統特點

框架包含高級日誌系統:

1. 非阻塞寫入: 使用隊列處理器將日誌寫入操作移至背景線程
2. 自動堆疊追蹤: 增強型TracebackLogger自動添加詳細錯誤信息
3. 全局異常捕獲: 捕獲並記錄所有未處理的異常，包括子線程異常
4. 進程感知: 適當處理多進程環境中的日誌初始化和配置
5. 按級別鎖定: 不同日誌級別使用獨立鎖，減少寫入競爭

📄 授權

本專案採用 MIT 授權條款。

🙏 致謝

感謝所有貢獻者和測試者。如有問題或建議，請提交 Issue 或 Pull Request。