Python如何批量合并表格

发布时间：2021-11-25 11:31:36 作者：iii
来源：亿速云阅读：354

# Python如何批量合并表格

在日常数据处理工作中，我们经常需要将多个结构相似的表格文件合并为一个统一的数据集。本文将详细介绍使用Python实现表格批量合并的5种实用方法，涵盖不同场景下的解决方案。

## 一、准备工作

### 1.1 安装必要库
```python
pip install pandas openpyxl xlrd

1.2 示例文件结构

假设我们有如下CSV文件需要合并：

data/
   ├── sales_Q1.csv
   ├── sales_Q2.csv
   └── sales_Q3.csv

二、基础合并方法

2.1 使用pandas.concat合并CSV

import pandas as pd
import glob

# 获取所有CSV文件
file_paths = glob.glob("data/*.csv")

# 读取并合并数据
dfs = []
for file in file_paths:
    df = pd.read_csv(file)
    dfs.append(df)

result = pd.concat(dfs, ignore_index=True)
result.to_csv("merged_sales.csv", index=False)

参数说明： - ignore_index=True：重置行索引 - axis=0：纵向堆叠（默认） - join='outer'：保留所有列（默认）

2.2 处理表头不一致的情况

# 方法1：统一使用第一个文件的表头
header = pd.read_csv(file_paths[0]).columns
result = pd.concat([pd.read_csv(f, names=header) for f in file_paths])

# 方法2：自动对齐不同列
result = pd.concat([pd.read_csv(f) for f in file_paths], join='outer')

三、进阶合并技巧

3.1 合并Excel多个工作表

with pd.ExcelWriter("merged.xlsx") as writer:
    for i, file in enumerate(file_paths):
        pd.read_csv(file).to_excel(writer, sheet_name=f"Sheet{i+1}")

3.2 添加来源标识列

dfs = []
for file in file_paths:
    df = pd.read_csv(file)
    df['source_file'] = file.split('/')[-1]  # 添加来源列
    dfs.append(df)

3.3 处理大型数据集

# 分块读取
chunks = []
for file in file_paths:
    for chunk in pd.read_csv(file, chunksize=10000):
        chunks.append(chunk)
result = pd.concat(chunks)

四、特殊场景处理

4.1 合并不同格式文件

readers = {
    '.csv': pd.read_csv,
    '.xlsx': pd.read_excel,
    '.json': pd.read_json
}

dfs = []
for file in file_paths:
    ext = file[file.rfind('.'):]
    dfs.append(readers[ext](file))

4.2 基于关键列合并

result = dfs[0]
for df in dfs[1:]:
    result = result.merge(df, on='product_id', how='outer')

4.3 处理编码问题

encodings = ['utf-8', 'gbk', 'latin1']
for file in file_paths:
    for enc in encodings:
        try:
            df = pd.read_csv(file, encoding=enc)
            break
        except UnicodeDecodeError:
            continue

五、性能优化方案

5.1 使用多线程加速

from concurrent.futures import ThreadPoolExecutor

def read_file(file):
    return pd.read_csv(file)

with ThreadPoolExecutor() as executor:
    dfs = list(executor.map(read_file, file_paths))

5.2 内存优化技巧

# 指定数据类型减少内存占用
dtypes = {'price': 'float32', 'quantity': 'int16'}
pd.read_csv(file, dtype=dtypes)

5.3 使用Dask处理超大数据

import dask.dataframe as dd

df = dd.read_csv('data/*.csv')
df.compute().to_csv('merged.csv')

六、完整实战案例

6.1 企业销售数据合并

import os
from datetime import datetime

def merge_sales_data(input_dir, output_file):
    """
    合并销售数据并添加处理元数据
    """
    dfs = []
    meta = {
        'merge_time': datetime.now().strftime('%Y-%m-%d %H:%M'),
        'source_files': []
    }
    
    for file in os.listdir(input_dir):
        if file.endswith(('.csv', '.xlsx')):
            try:
                path = os.path.join(input_dir, file)
                if file.endswith('.csv'):
                    df = pd.read_csv(path)
                else:
                    df = pd.read_excel(path)
                
                df['file_source'] = file
                dfs.append(df)
                meta['source_files'].append(file)
                
            except Exception as e:
                print(f"Error processing {file}: {str(e)}")
    
    if not dfs:
        raise ValueError("No valid files found")
    
    result = pd.concat(dfs, ignore_index=True)
    
    # 保存合并结果
    if output_file.endswith('.csv'):
        result.to_csv(output_file, index=False)
    else:
        result.to_excel(output_file, index=False)
    
    # 保存元数据
    with open('merge_meta.json', 'w') as f:
        json.dump(meta, f)
    
    return result.shape

6.2 自动化监控脚本

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class CSVHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.src_path.endswith('.csv'):
            merge_sales_data('input_dir', 'merged.csv')

observer = Observer()
observer.schedule(CSVHandler(), path='input_dir')
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

七、常见问题解答

Q1: 如何处理合并时的重复数据？

# 方法1：标记重复项
result['is_duplicate'] = result.duplicated()

# 方法2：删除完全重复的行
result.drop_duplicates(inplace=True)

# 方法3：基于关键列去重
result.drop_duplicates(subset=['order_id'], keep='last')

Q2: 合并后如何保持原始顺序？

# 添加原始序号列
for i, df in enumerate(dfs):
    df['original_order'] = range(1, len(df)+1)
    df['batch_num'] = i+1

result = pd.concat(dfs).sort_values(['batch_num', 'original_order'])

Q3: 合并超大型文件时的内存管理？

使用chunksize参数分块读取
考虑使用数据库作为中间存储
使用dtype参数优化数据类型
定期释放不用的变量：del df; gc.collect()

八、总结

本文介绍了Python批量合并表格的完整方案，关键要点包括： 1. 基础合并使用pd.concat() 2. 复杂合并考虑pd.merge() 3. 大型数据集采用分块处理 4. 多格式文件需要类型判断 5. 生产环境应添加错误处理和日志记录

根据实际需求选择合适的方法，通常： - 简单合并：方法一 - 结构化合并：方法四 - 超大数据：方法五

扩展学习： - pandas官方文档 - Python数据清洗实战 - 高效Pandas技巧 “`