1. 为什么你需要掌握Python gzip库？
在日常开发中，我们经常会遇到需要处理大文件的情况。比如服务器日志、数据备份、或者从API获取的压缩数据。这时候，gzip就像是一个神奇的文件瘦身专家，能把文件体积压缩到原来的1/3甚至更小。我最近处理一个2GB的日志文件时，用gzip压缩后只剩600MB，不仅节省了存储空间，传输速度也快了不少。
Python内置的gzip模块用起来特别顺手，不需要安装任何第三方库。它实际上是基于GNU zip算法的标准实现，在Linux/macOS上都能无缝使用。Windows用户也完全不用担心兼容性问题，我在Win10和Win11上都测试过，表现非常稳定。
这个库特别适合处理文本类文件，比如：
- 日志文件（.log）
- 配置文件（.json, .yaml）
- 数据备份文件
- API返回的压缩数据
2. 基础操作：从单个文件开始
2.1 创建并压缩新文件
我们先从最简单的场景开始：创建一个全新的gzip压缩文件。这里有个坑我踩过，直接写入字符串会报错，必须先把内容编码成bytes。
import gzip content = "这是要压缩的内容，可以是任意文本" with gzip.open('example.gz', 'wb') as f: f.write(content.encode('utf-8')) # 关键步骤：编码为bytes实际项目中，我建议总是使用with语句来操作文件。这样即使程序中途崩溃，文件也会正确关闭，不会损坏压缩包。有一次我忘了用with，结果压缩到一半程序异常退出，生成了一个无法打开的损坏文件。
2.2 读取压缩文件内容
读取gzip文件就像读取普通文件一样简单，但要注意解码：
with gzip.open('example.gz', 'rb') as f: compressed_data = f.read() # 获取二进制数据 text_content = compressed_data.decode('utf-8') # 解码为字符串 print(text_content)这里有个实用技巧：如果你不确定文件编码，可以先尝试'utf-8'，如果失败再尝试其他编码。我在处理不同来源的日志文件时，经常遇到编码问题，通常会这样处理：
encodings = ['utf-8', 'gbk', 'latin-1'] for enc in encodings: try: print(compressed_data.decode(enc)) break except UnicodeDecodeError: continue2.3 压缩现有文件
更常见的场景是压缩已经存在的文件。这里我分享一个更健壮的版本，加入了错误处理：
import gzip import os def compress_file(input_path, output_path): if not os.path.exists(input_path): raise FileNotFoundError(f"输入文件不存在: {input_path}") try: with open(input_path, 'rb') as f_in: with gzip.open(output_path, 'wb') as f_out: f_out.writelines(f_in) print(f"成功压缩: {input_path} -> {output_path}") return True except Exception as e: print(f"压缩失败: {e}") return False # 使用示例 compress_file('large_log.txt', 'compressed_log.gz')这个函数会返回操作是否成功，方便在自动化脚本中判断后续操作。我在一个日志收集系统中就用了类似的实现，每天凌晨自动压缩前一天的日志。
3. 进阶技巧：批量处理与自动化
3.1 批量压缩文件夹
当需要处理多个文件时，手动一个个操作就太麻烦了。这是我常用的批量压缩函数：
import os import gzip from pathlib import Path def batch_compress(directory, extension='.txt'): """压缩目录下所有指定扩展名的文件""" base_dir = Path(directory) for file_path in base_dir.glob(f'*{extension}'): output_path = file_path.with_suffix(file_path.suffix + '.gz') try: with open(file_path, 'rb') as f_in: with gzip.open(output_path, 'wb') as f_out: f_out.writelines(f_in) print(f"压缩成功: {file_path.name}") # 可选：删除原文件 # file_path.unlink() except Exception as e: print(f"压缩失败 {file_path.name}: {e}") # 使用示例 batch_compress('/var/log/myapp', '.log')这个函数用到了pathlib，这是Python 3.4+引入的更现代的文件路径操作方式。比起传统的os.path，pathlib的API更加直观。我在项目中全面转向使用pathlib后，代码可读性提高了很多。
3.2 定时自动压缩脚本
结合Python的schedule库，可以轻松实现定时压缩：
import schedule import time from datetime import datetime def job(): print(f"{datetime.now()}: 开始执行压缩任务") batch_compress('/var/log/myapp', '.log') print(f"{datetime.now()}: 压缩任务完成") # 每天凌晨2点执行 schedule.every().day.at("02:00").do(job) while True: schedule.run_pending() time.sleep(60)这个脚本可以放在后台运行，或者更好的是配置为系统服务。我在Linux服务器上用的是systemd来管理这类Python脚本，确保它们能自动重启。
4. 性能优化与实用技巧
4.1 压缩级别调整
gzip支持不同的压缩级别，从0（不压缩）到9（最大压缩）。默认是6，在速度和压缩率之间取得平衡。根据我的测试：
| 级别 | 压缩时间 | 文件大小 | 适用场景 |
|---|---|---|---|
| 1 | 最快 | 较大 | 需要快速压缩 |
| 6 | 中等 | 中等 | 日常使用（默认） |
| 9 | 最慢 | 最小 | 需要最小体积 |
使用方法：
import gzip import shutil with open('big_file.txt', 'rb') as f_in: with gzip.open('compressed.gz', 'wb', compresslevel=9) as f_out: shutil.copyfileobj(f_in, f_out)注意：更高的压缩级别会显著增加CPU使用率。在批量处理大量文件时，建议先在测试环境评估性能影响。
4.2 内存高效处理大文件
处理超大文件时，一次性读取整个文件会消耗大量内存。更安全的方式是分块处理：
def compress_large_file(input_path, output_path, chunk_size=1024*1024): """分块压缩大文件，避免内存不足""" with open(input_path, 'rb') as f_in: with gzip.open(output_path, 'wb') as f_out: while True: chunk = f_in.read(chunk_size) # 每次读取1MB if not chunk: break f_out.write(chunk)这个函数在我的一个数据分析项目中帮了大忙，当时需要处理20GB+的CSV文件，普通方法直接导致内存溢出。
4.3 与其他工具配合使用
gzip经常和其他命令行工具配合使用。比如在Python中调用系统命令：
import subprocess # 使用系统gzip工具（通常比Python实现更快） subprocess.run(['gzip', '-k', 'large_file.txt']) # 解压并直接处理内容 process = subprocess.Popen(['gunzip', '-c', 'data.gz'], stdout=subprocess.PIPE) for line in process.stdout: print(line.decode().strip())这种方法在需要最高性能时特别有用，因为系统级的gzip实现通常经过高度优化。
5. 实际应用案例：日志处理系统
最后分享一个我在实际项目中实现的日志处理方案。这个系统每天自动：
- 压缩前一天的日志
- 检查磁盘空间，自动清理旧日志
- 发送压缩日志到远程备份
import gzip import os from datetime import datetime, timedelta def manage_logs(log_dir, keep_days=30): today = datetime.now() cutoff = today - timedelta(days=keep_days) for filename in os.listdir(log_dir): if not filename.endswith('.log'): continue filepath = os.path.join(log_dir, filename) file_time = datetime.fromtimestamp(os.path.getmtime(filepath)) # 压缩非当天的日志 if file_time.date() < today.date(): compress_log(filepath) # 删除过期的压缩日志 if filename.endswith('.gz') and file_time < cutoff: os.remove(filepath) def compress_log(log_path): gz_path = f"{log_path}.gz" try: with open(log_path, 'rb') as f_in: with gzip.open(gz_path, 'wb') as f_out: f_out.writelines(f_in) os.remove(log_path) # 压缩成功后删除原文件 return True except Exception as e: print(f"压缩失败 {log_path}: {e}") return False这个脚本可以设置为每天运行一次，我通常用cronjob来调度。它帮我们节省了70%的日志存储空间，同时保证了日志的可追溯性。
