优化大规模CSV文件读取：解决Pandas与XGBoost内存问题的策略

import xgboost as xgb
import pandas as pd
from typing import List

# 假设您有一个自定义的迭代器类，用于分块读取CSV文件
class CSVBatchIterator(xgb.DataIter):
    def __init__(self, file_paths: List[str], batch_size: int):
        super().__init__()
        self.file_paths = file_paths
        self.batch_size = batch_size
        self.it = iter(self._read_chunks())
        self.data = None
        self.labels = None

    def _read_chunks(self):
        # 这是一个简化示例，实际可能需要更复杂的逻辑来处理跨文件分块
        for path in self.file_paths:
            df = pd.read_csv(path)
            # 假设最后一列是标签，其余是特征
            features = df.iloc[:, :-1]
            labels = df.iloc[:, -1]

            # 进一步将单个文件的DataFrame分块
            for i in range(0, len(features), self.batch_size):
                yield features.iloc[i:i+self.batch_size], labels.iloc[i:i+self.batch_size]

    def next(self, input_data):
        try:
            features, labels = next(self.it)
            self.data = features
            self.labels = labels
            input_data(data=self.data, label=self.labels)
            return 1 # Return 1 for successful iteration
        except StopIteration:
            return 0 # Return 0 for end of iteration

# 示例使用
# file_paths = ["path/to/file1.csv", "path/to/file2.csv", ...]
# batch_iterator = CSVBatchIterator(file_paths, batch_size=100000)
# dtrain = xgb.DMatrix(batch_iterator)
#
# # 训练模型
# # params = {'objective': 'binary:logistic', 'eval_metric': 'logloss'}
# # bst = xgb.train(params, dtrain)

import pandas as pd
import multiprocessing as mp
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_EXCEPTION
from typing import List
import logging

logger = logging.getLogger(__name__)

def _read_training_data(training_data_path: str) -> pd.DataFrame:
    """
    单个CSV文件读取函数。
    """
    df = pd.read_csv(training_data_path)
    return df

def read_training_data_optimized(
    paths: List[str]
) -> pd.DataFrame:
    """
    优化后的并行读取训练数据函数。