Python 3.8 CSV 数据处理实战2012-2019 体育收入榜单的 3 种分析函数在数据分析的日常工作中CSV 文件处理是最基础却最频繁遇到的任务之一。体育产业作为全球增长最快的领域之一其收入数据蕴含着丰富的商业价值。本文将基于 2012-2019 年体育收入榜单数据通过三个精心设计的 Python 函数展示从基础读取到高级分析的完整数据处理流程。1. 数据读取与预处理数据读取是分析的起点也是容易出错的关键环节。我们首先构建一个健壮的read_file函数它不仅完成基础的文件读取还内置了数据清洗逻辑。def read_file(file_path2012-19sport.csv): 读取CSV文件并进行初步清洗 :param file_path: 文件路径默认为2012-19sport.csv :return: 清洗后的数据列表 processed_data [] try: with open(file_path, r, encodingutf-8) as data_file: # 跳过标题行 data_file.readline() for line in data_file: # 去除前后空格并按逗号分割 row line.strip().split(,) # 清洗特殊字符 row[0] row[0].replace(#, ) processed_data.append(row) except FileNotFoundError: print(f错误文件 {file_path} 未找到) except Exception as e: print(f读取文件时发生错误{str(e)}) return processed_data这个改进版本增加了以下关键特性异常处理捕获文件不存在和其他潜在错误参数化文件路径提高函数复用性清晰的文档字符串说明函数用途和参数更简洁的变量命名增强代码可读性提示在实际项目中建议使用csv模块代替直接的文件操作它能更好地处理CSV格式的特殊情况如包含逗号的字段。2. 年度收入排名分析fun1函数负责按年份筛选并展示收入最高的前K条记录。我们将其重构为更专业的analyze_top_earnings函数def analyze_top_earnings(data, year, top_k10): 分析指定年份收入最高的前K条记录 :param data: 数据集 :param year: 目标年份 :param top_k: 显示前多少条记录默认为10 :return: 无返回值直接打印结果 # 筛选指定年份的数据 yearly_data [row for row in data if row[6] str(year)] if not yearly_data: print(f警告未找到 {year} 年的数据) return # 按收入排序降序 yearly_data.sort(keylambda x: float(x[2].replace($, ).replace( M, )), reverseTrue) # 打印表头 print(f\n{year} 年收入前 {top_k} 名:) print(- * 50) print(排名 | 名称 | 收入 | 运动项目 | 年份) print(- * 50) # 打印结果 for idx, item in enumerate(yearly_data[:top_k], 1): print(f{idx:2d} | {item[0]:20s} | {item[2]:8s} | {item[5]:15s} | {item[6]})改进亮点包括默认参数设置合理的默认显示数量数据验证检查空结果并给出友好提示格式化输出美观的表格形式展示性能优化使用列表推导式和lambda表达式3. 运动项目收入分析原始fun2函数功能较为复杂我们将其拆分为两个更专注的函数list_sports_by_year和analyze_sport_earnings。3.1 列出指定年份的运动项目def list_sports_by_year(data, year): 列出指定年份包含的所有运动项目 :param data: 数据集 :param year: 目标年份 :return: 运动项目列表 yearly_data [row for row in data if row[6] str(year)] sports sorted(list(set(row[5] for row in yearly_data))) print(f\n{year} 年包含的运动项目:) print(- * 30) for idx, sport in enumerate(sports, 1): print(f{idx:2d}. {sport}) return sports3.2 分析特定运动项目的收入def analyze_sport_earnings(data, year, sport_index, sports_list): 分析指定年份特定运动项目的收入详情 :param data: 数据集 :param year: 目标年份 :param sport_index: 运动项目索引 :param sports_list: 运动项目列表 :return: 无返回值直接打印结果 try: selected_sport sports_list[sport_index - 1] except IndexError: print(错误无效的运动项目编号) return # 筛选数据 sport_data [row for row in data if row[6] str(year) and row[5] selected_sport] if not sport_data: print(f警告未找到 {year} 年 {selected_sport} 的数据) return # 计算总收入 total sum(float(row[2].replace($, ).replace( M, )) for row in sport_data) # 打印结果 print(f\n{year} 年 {selected_sport} 收入详情:) print(- * 60) print(名称 | 收入 | 排名 | 年份) print(- * 60) for row in sport_data: print(f{row[0]:25s} | {row[2]:8s} | {row[0]:4s} | {row[6]}) print(- * 60) print(f总收入: ${total:.2f} M)4. 构建完整的数据处理工具将上述函数整合为一个可复用的工具类class SportsEarningsAnalyzer: def __init__(self, file_path): self.data self._load_data(file_path) def _load_data(self, file_path): 私有方法加载并清洗数据 # 实现与之前read_file类似的逻辑 pass def analyze_by_year(self, year, top_k10): 分析指定年份的收入排名 # 实现与analyze_top_earnings类似的逻辑 pass def analyze_sports(self, year): 分析指定年份的运动项目 sports self.list_sports_by_year(year) while True: try: choice int(input(请输入要分析的运动项目编号(0退出): )) if choice 0: break self.analyze_sport_earnings(year, choice, sports) except ValueError: print(请输入有效的数字) # 其他方法实现...这个类提供了更结构化的接口支持以下功能数据封装将数据和处理逻辑绑定在一起交互式界面方便用户逐步探索数据可扩展性易于添加新的分析方法5. 数据清洗最佳实践在处理真实数据时清洗是必不可少的步骤。以下是三个关键实践统一货币格式处理def normalize_currency(value): 统一处理货币字符串 :param value: 原始字符串如$ 12.34 M :return: 浮点数值 try: return float(value.replace($, ).replace( M, ).strip()) except ValueError: print(f警告无法解析货币值 {value}) return 0.0处理缺失数据def handle_missing_data(data, default_values): 处理数据集中的缺失值 :param data: 原始数据集 :param default_values: 各列的默认值字典 :return: 清洗后的数据集 cleaned_data [] for row in data: cleaned_row [] for idx, value in enumerate(row): if not value.strip(): # 使用默认值填充 cleaned_row.append(str(default_values.get(idx, ))) else: cleaned_row.append(value) cleaned_data.append(cleaned_row) return cleaned_data数据验证def validate_data(data, expected_columns): 验证数据基本完整性 :param data: 数据集 :param expected_columns: 预期的列数 :return: 布尔值表示数据是否有效 if not data: print(错误数据集为空) return False for row in data: if len(row) ! expected_columns: print(f警告发现异常行预期 {expected_columns} 列实际 {len(row)} 列) print(row) return False return True6. 性能优化技巧处理大型数据集时性能变得至关重要。以下是几个优化建议使用生成器处理大文件def read_large_file(file_path): 逐行读取大文件节省内存 with open(file_path, r, encodingutf-8) as f: # 跳过标题 next(f) for line in f: yield line.strip().split(,)使用pandas进行高效分析import pandas as pd def analyze_with_pandas(file_path): 使用pandas进行高效数据分析 df pd.read_csv(file_path) # 清洗数据 df[收入] df[收入].str.replace($, ).str.replace( M, ).astype(float) # 示例分析按年份和运动项目分组汇总 result df.groupby([年份, 运动项目])[收入].agg([sum, mean, count]) return result缓存常用计算结果from functools import lru_cache lru_cache(maxsize32) def get_year_data(year): 缓存年度数据查询结果 # 实现数据查询逻辑 pass7. 实际应用案例让我们看一个完整的应用示例展示如何使用这些工具分析体育收入数据def main(): # 初始化分析器 analyzer SportsEarningsAnalyzer(2012-19sport.csv) # 交互式菜单 while True: print(\n体育收入数据分析工具) print(1. 查看年度收入排名) print(2. 分析运动项目收入) print(3. 退出) choice input(请选择操作: ) if choice 1: year int(input(请输入年份(2012-2019): )) analyzer.analyze_by_year(year) elif choice 2: year int(input(请输入年份(2012-2019): )) analyzer.analyze_sports(year) elif choice 3: break else: print(无效选择请重新输入) if __name__ __main__: main()这个案例展示了如何将各个功能模块组合成一个完整的应用程序提供友好的用户界面。