Python 数据分析知识点总结
基于个人学习笔记整理，聚焦数据分析实战方向
一、Python 基础特性
Python 是一门解释型的弱类型动态语言，非常适合快速开发和数据分析。
- 解释型：边解释边运行，灵活易改，适合探索性分析
- 弱类型：变量无需声明类型，隐式转换
- 动态型：运行时确定数据类型，处理各种格式的数据非常方便
二、数据类型（数据分析核心）
2.1 字符串（str）- 数据清洗的基础
# f-string 格式化（推荐，最简洁）
name = '张三'
age = 25
print(f'{name}今年{age}岁')
# 常用字符串方法（数据清洗必备）
text = " Hello, World! "
text.strip() # 去除两端空格 → 'Hello, World!'
text.lower() # 转小写 → ' hello, world! '
text.upper() # 转大写 → ' HELLO, WORLD! '
# 分割与拼接（CSV解析常用）
data = "张三,25,北京"
parts = data.split(",") # ['张三', '25', '北京']
joined = "-".join(parts) # '张三-25-北京'
# 查找与替换（数据清洗）
log = "Error: 文件未找到"
log.replace("Error", "Warning") # 'Warning: 文件未找到'
log.find("文件") # 返回位置：7
log.count("未") # 统计次数：1
# 类型判断（数据验证）
"12345".isdigit() # True - 判断是否为纯数字
"hello".isalpha() # True - 判断是否为纯字母
2.2 列表（list）- 数据存储的核心
# 增删改查
data = [10, 20, 30, 40, 50]
# 增
data.append(60) # 末尾追加 → [10, 20, 30, 40, 50, 60]
data.extend([70, 80]) # 批量追加 → [10, 20, 30, 40, 50, 60, 70, 80]
data.insert(0, 5) # 指定位置插入 → [5, 10, 20, 30, 40, 50, 60, 70, 80]
# 删
data.remove(30) # 按值删除（第一个匹配项）
data.pop(0) # 按索引删除，返回被删除的值
data.pop() # 默认删除最后一个
del data[0:2] # 删除切片
# 查
data = [10, 20, 30, 40, 50]
data.index(30) # 返回索引：2
30 in data # 成员判断：True
data.count(30) # 统计次数：1
# 排序
scores = [85, 92, 78, 90, 88]
sorted(scores) # 升序，返回新列表
sorted(scores, reverse=True) # 降序，返回新列表
scores.sort() # 原地升序排序
scores.sort(reverse=True) # 原地降序排序
# 自定义排序（数据分析常用）
students = [
{"name": "张三", "score": 85},
{"name": "李四", "score": 92},
{"name": "王五", "score": 78}
]
sorted(students, key=lambda x: x["score"], reverse=True) # 按成绩降序
2.3 字典（dict）- 结构化数据的关键
# 创建
person = {"name": "张三", "age": 25, "city": "北京"}
# 查
person.get("name") # '张三'（推荐，不存在返回None）
person.get("phone", "无") # '无'（设置默认值）
person.keys() # 获取所有键
person.values() # 获取所有值
person.items() # 获取所有键值对
# 增改
person["email"] = "zhangsan@example.com" # 新增
person["age"] = 26 # 修改
person.update({"age": 27, "phone": "138xxxx"}) # 批量更新
# 删
person.pop("email") # 删除指定键值对，返回值
del person["phone"] # 删除指定键值对
# 遍历（数据分析常用）
for key, value in person.items():
print(f"{key}: {value}")
2.4 集合（set）- 数据去重与集合运算
# 核心用途：去重
data = [1, 2, 2, 3, 3, 3, 4, 4, 4, 4]
unique_data = list(set(data)) # [1, 2, 3, 4]
# 集合运算（数据分析中用于对比两组数据）
set_a = {1, 2, 3, 4, 5}
set_b = {4, 5, 6, 7, 8}
set_a & set_b # 交集 → {4, 5}
set_a | set_b # 并集 → {1, 2, 3, 4, 5, 6, 7, 8}
set_a - set_b # 差集 → {1, 2, 3}
set_a ^ set_b # 对称差集 → {1, 2, 3, 6, 7, 8}
2.5 元组（tuple）- 不可变数据
# 特点：不可修改，常用于函数返回多个值
coordinates = (116.4, 39.9) # 北京坐标
# 单元素元组必须加逗号
single = (42,) # 这是元组
not_tuple = (42) # 这是整数
# 解包（数据分析常用）
x, y = coordinates
三、类型转换（数据处理必备）
# 基本类型转换
int("42") # 字符串转整数 → 42
int(3.14) # 浮点转整数 → 3（截断，不是四舍五入）
float("3.14") # 字符串转浮点 → 3.14
str(12345) # 数字转字符串 → '12345'
# 集合类型转换
list("hello") # 字符串转列表 → ['h', 'e', 'l', 'l', 'o']
tuple([1, 2, 3]) # 列表转元组 → (1, 2, 3)
set([1, 2, 2, 3]) # 列表转集合（去重）→ {1, 2, 3}
# 字典转换（数据分析常用）
keys = ['name', 'age', 'city']
values = ['张三', 25, '北京']
dict(zip(keys, values)) # {'name': '张三', 'age': 25, 'city': '北京'}
# 安全转换（处理脏数据）
def safe_int(value, default=0):
try:
return int(value)
except (ValueError, TypeError):
return default
safe_int("123") # 123
safe_int("abc") # 0
safe_int(None) # 0
四、推导式（高效数据处理）
4.1 列表推导式
# 语法：[表达式 for 变量 in 可迭代对象 if 条件]
# 生成平方数列表
squares = [x**2 for x in range(1, 11)]
# [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
# 筛选偶数
evens = [x for x in range(1, 21) if x % 2 == 0]
# [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]
# 数据清洗示例：去除空字符串
raw_data = ["张三", "", "李四", " ", "王五"]
clean_data = [x.strip() for x in raw_data if x.strip()]
# ['张三', '李四', '王五']
4.2 字典推导式
# 语法：{键:值 for 变量 in 可迭代对象 if 条件}
# 创建平方数字典
square_dict = {x: x**2 for x in range(1, 6)}
# {1: 1, 2: 4, 3: 9, 4: 16, 5: 25}
# 数据转换示例：将列表转为字典
names = ['张三', '李四', '王五']
scores = [85, 92, 78]
score_dict = {name: score for name, score in zip(names, scores)}
# {'张三': 85, '李四': 92, '王五': 78}
4.3 集合推导式
# 语法：{表达式 for 变量 in 可迭代对象 if 条件}
# 去重并转换为集合
data = [1, 2, 2, 3, 3, 3, 4, 4, 4, 4]
unique_set = {x for x in data}
# {1, 2, 3, 4}
五、函数（数据处理逻辑封装）
5.1 基本函数定义
# 位置参数
def calculate_average(scores):
"""计算平均分"""
return sum(scores) / len(scores)
# 默认参数
def format_number(num, decimal_places=2):
"""格式化数字，保留指定小数位"""
return round(num, decimal_places)
# 可变参数（处理不定长数据）
def sum_all(*args):
"""计算所有参数的和"""
return sum(args)
sum_all(1, 2, 3, 4, 5) # 15
# 关键字参数（灵活配置）
def create_report(title, **kwargs):
"""创建报告，支持自定义属性"""
report = {"title": title}
report.update(kwargs)
return report
create_report("销售报告", author="张三", date="2024-01-01")
5.2 匿名函数（lambda）
# 语法：lambda 参数: 表达式
# 常用于排序和数据转换
square = lambda x: x ** 2
square(5) # 25
# 排序场景
students = [
{"name": "张三", "age": 20},
{"name": "李四", "age": 18},
{"name": "王五", "age": 22}
]
sorted(students, key=lambda s: s["age"]) # 按年龄升序
5.3 内置函数（数据分析常用）
# 数学计算
abs(-5) # 绝对值 → 5
round(3.14159, 2) # 四舍五入 → 3.14
sum([1, 2, 3, 4]) # 求和 → 10
min([5, 3, 8, 1]) # 最小值 → 1
max([5, 3, 8, 1]) # 最大值 → 8
divmod(17, 5) # 商和余数 → (3, 2)
# 序列操作
len([1, 2, 3]) # 长度 → 3
sorted([3, 1, 2]) # 排序 → [1, 2, 3]
reversed([1, 2, 3]) # 反转迭代器
enumerate(['a', 'b', 'c']) # 带索引遍历
zip([1, 2], ['a', 'b']) # 配对
# 高级函数
map(lambda x: x**2, [1, 2, 3]) # 映射 → [1, 4, 9]
filter(lambda x: x > 2, [1, 2, 3, 4]) # 过滤 → [3, 4]
# 类型检查
isinstance(42, int) # True
isinstance("hello", str) # True
type(42) # <class 'int'>
六、文件操作（数据读写）
6.1 文件读写基础
# 读取文件
with open("data.txt", "r", encoding="utf-8") as f:
content = f.read() # 读取全部内容
# 或逐行读取
for line in f:
print(line.strip())
# 写入文件
with open("output.txt", "w", encoding="utf-8") as f:
f.write("第一行\n")
f.write("第二行\n")
# 追加写入
with open("log.txt", "a", encoding="utf-8") as f:
f.write("新日志内容\n")
6.2 文件打开模式
| 模式 | 说明 | 模式 | 说明 |
|---|---|---|---|
r | 只读（默认） | rb | 二进制只读 |
w | 写入（覆盖） | wb | 二进制写入 |
a | 追加写入 | ab | 二进制追加 |
r+ | 读写 | rb+ | 二进制读写 |
6.3 JSON 数据处理（API 数据常用）
import json
# Python对象 → JSON字符串
data = {"name": "张三", "age": 25, "scores": [85, 92, 78]}
json_str = json.dumps(data, ensure_ascii=False)
print(json_str) # {"name": "张三", "age": 25, "scores": [85, 92, 78]}
# JSON字符串 → Python对象
parsed = json.loads(json_str)
print(parsed["name"]) # 张三
# 写入JSON文件
with open("data.json", "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# 读取JSON文件
with open("data.json", "r", encoding="utf-8") as f:
loaded_data = json.load(f)
6.4 CSV 数据处理
import csv
# 写入CSV
data = [
["姓名", "年龄", "城市"],
["张三", 25, "北京"],
["李四", 30, "上海"],
["王五", 28, "广州"]
]
with open("data.csv", "w", encoding="utf-8-sig", newline="") as f:
writer = csv.writer(f)
writer.writerows(data)
# 读取CSV
with open("data.csv", "r", encoding="utf-8") as f:
reader = csv.reader(f)
for row in reader:
print(row)
七、常用模块（数据分析工具箱）
7.1 os 模块 - 文件系统操作
import os
os.getcwd() # 获取当前工作目录
os.listdir(".") # 列出当前目录文件
os.path.exists("data.csv") # 判断文件是否存在
os.path.isfile("data.csv") # 判断是否为文件
os.path.isdir("data") # 判断是否为目录
os.path.join("data", "file.csv") # 拼接路径
os.path.split("/data/file.csv") # 拆分路径 → ('/data', 'file.csv')
os.path.splitext("file.csv") # 拆分扩展名 → ('file', '.csv')
os.mkdir("new_folder") # 创建目录
os.makedirs("a/b/c") # 递归创建目录
os.remove("file.txt") # 删除文件
os.rename("old.txt", "new.txt") # 重命名
7.2 json 模块 - 结构化数据
（已在文件操作部分介绍）
7.3 time 模块 - 时间处理
import time
time.time() # 当前时间戳
time.localtime() # 本地时间结构体
time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) # 格式化时间
time.sleep(1) # 暂停1秒
# 常用格式化符号
# %Y: 四位年份 %m: 月份 %d: 日期
# %H: 24小时制 %M: 分钟 %S: 秒
7.4 random 模块 - 随机数据生成
import random
random.random() # [0, 1) 随机浮点数
random.randint(1, 100) # [1, 100] 随机整数
random.uniform(1.0, 10.0) # [1.0, 10.0] 随机浮点数
random.choice([1, 2, 3, 4]) # 随机选择一个元素
random.sample([1, 2, 3, 4, 5], 3) # 随机选择3个不重复元素
random.shuffle([1, 2, 3, 4]) # 原地打乱列表
7.5 hashlib 模块 - 数据校验
import hashlib
# 计算字符串哈希值
data = "Hello, World!"
md5_hash = hashlib.md5(data.encode()).hexdigest()
print(f"MD5: {md5_hash}")
sha256_hash = hashlib.sha256(data.encode()).hexdigest()
print(f"SHA256: {sha256_hash}")
# 计算文件哈希值（校验文件完整性）
def file_hash(filepath, algorithm="md5"):
h = hashlib.new(algorithm)
with open(filepath, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
h.update(chunk)
return h.hexdigest()
八、异常处理（健壮的数据处理）
# 基本结构
try:
result = 10 / 0
except ZeroDivisionError as e:
print(f"错误: {e}")
except Exception as e:
print(f"未知错误: {e}")
else:
print("没有异常发生")
finally:
print("无论如何都会执行")
# 实际应用：安全读取数据
def safe_read_json(filepath):
try:
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
except FileNotFoundError:
print(f"文件不存在: {filepath}")
return None
except json.JSONDecodeError:
print(f"JSON格式错误: {filepath}")
return None
except Exception as e:
print(f"读取失败: {e}")
return None
# 自定义异常
class DataFormatError(Exception):
"""数据格式异常"""
pass
def validate_age(age):
if not isinstance(age, int) or age < 0 or age > 150:
raise DataFormatError(f"无效的年龄: {age}")
return True
九、面向对象基础（数据模型封装）
9.1 类与对象
class Student:
"""学生类"""
def __init__(self, name, age, scores):
self.name = name
self.age = age
self.scores = scores
def average_score(self):
"""计算平均分"""
return sum(self.scores) / len(self.scores)
def __str__(self):
return f"Student(name={self.name}, age={self.age})"
# 创建对象
student = Student("张三", 20, [85, 92, 78])
print(student) # Student(name=张三, age=20)
print(student.average_score()) # 85.0
9.2 继承
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
def introduce(self):
return f"我叫{self.name}，今年{self.age}岁"
class Student(Person):
def __init__(self, name, age, student_id):
super().__init__(name, age) # 调用父类构造方法
self.student_id = student_id
def introduce(self):
return f"{super().introduce()}，学号{self.student_id}"
student = Student("张三", 20, "2024001")
print(student.introduce()) # 我叫张三，今年20岁，学号2024001
9.3 封装（私有属性）
class BankAccount:
def __init__(self, balance=0):
self.__balance = balance # 私有属性
@property
def balance(self):
"""获取余额"""
return self.__balance
@balance.setter
def balance(self, amount):
"""设置余额（带验证）"""
if amount < 0:
raise ValueError("余额不能为负数")
self.__balance = amount
account = BankAccount(1000)
print(account.balance) # 1000
account.balance = 2000 # 正常
# account.balance = -100 # 抛出异常
十、实战技巧汇总
10.1 数据清洗常用模式
# 1. 去除空值和空白
data = ["张三", "", " ", "李四", None]
clean = [x.strip() for x in data if x and x.strip()]
# 2. 类型转换容错
def safe_float(value, default=0.0):
try:
return float(value)
except (ValueError, TypeError):
return default
# 3. 数据去重（保持顺序）
def deduplicate(data):
seen = set()
return [x for x in data if not (x in seen or seen.add(x))]
# 4. 扁平化嵌套列表
def flatten(lst):
return [item for sublist in lst for item in sublist]
nested = [[1, 2], [3, 4], [5, 6]]
flatten(nested) # [1, 2, 3, 4, 5, 6]
10.2 数据统计常用模式
from collections import Counter
# 统计元素出现次数
data = ["苹果", "香蕉", "苹果", "橙子", "香蕉", "苹果"]
counter = Counter(data)
print(counter.most_common(2)) # [('苹果', 3), ('香蕉', 2)]
# 分组统计
students = [
{"name": "张三", "grade": "A"},
{"name": "李四", "grade": "B"},
{"name": "王五", "grade": "A"},
]
groups = {}
for s in students:
groups.setdefault(s["grade"], []).append(s["name"])
# {'A': ['张三', '王五'], 'B': ['李四']}
10.3 文件批量处理
import os
import glob
# 获取所有CSV文件
csv_files = glob.glob("data/*.csv")
# 批量读取并合并
all_data = []
for filepath in csv_files:
with open(filepath, "r", encoding="utf-8") as f:
reader = csv.reader(f)
header = next(reader) # 跳过表头
for row in reader:
all_data.append(row)
print(f"共读取 {len(all_data)} 条记录")
附录：编码规范速查
| 项目 | 规范 |
|---|---|
| 变量名 | snake_case（小写+下划线） |
| 类名 | PascalCase（首字母大写） |
| 常量 | UPPER_CASE（全大写） |
| 私有属性 | _protected（单下划线）或 __private（双下划线） |
| 缩进 | 4个空格 |
| 行长度 | 不超过80字符 |
| 导入顺序 | 内置 → 第三方 → 自定义 |
十一、正则表达式（数据清洗利器）
正则表达式是大规模数据清洗的核心工具，用于文本模式匹配、提取和替换。
11.1 基础语法速查
import re
# 常用元字符
# . 匹配任意字符（除换行）
# \d 匹配数字 [0-9]
# \w 匹配字母数字下划线 [a-zA-Z0-9_]
# \s 匹配空白字符
# \b 匹配单词边界
# ^ 匹配开头
# $ 匹配结尾
# 量词
# * 0次或多次
# + 1次或多次
# ? 0次或1次
# {n} 恰好n次
# {n,m} n到m次
11.2 核心函数
import re
text = "联系电话：13812345678，邮箱：test@example.com"
# re.search() - 查找第一个匹配
match = re.search(r"1[3-9]\d{9}", text)
if match:
print(match.group()) # 13812345678
# re.findall() - 查找所有匹配
phones = re.findall(r"1[3-9]\d{9}", "13812345678 15987654321")
print(phones) # ['13812345678', '15987654321']
# re.sub() - 替换
cleaned = re.sub(r"\d{3}\*{4}\d{4}", "***", "手机：138****5678")
# re.split() - 分割
parts = re.split(r"[;；,，]+", "Python;Java;C++")
print(parts) # ['Python', 'Java', 'C++']
# re.compile() - 预编译（大量匹配时性能更好）
pattern = re.compile(r"1[3-9]\d{9}")
result = pattern.findall("13812345678 15987654321")
11.3 分组提取
import re
# 命名分组
text = "2024-01-15"
match = re.search(r"(?P<year>\d{4})-(?P<month>\d{2})-(?P<day>\d{2})", text)
if match:
print(match.group("year")) # 2024
print(match.group("month")) # 01
# 非捕获分组
text = "hello123world456"
result = re.findall(r"(?:\d+)", text) # ['123', '456']
11.4 数据清洗实战模式
import re
# 1. 清理HTML标签
html = "<p>这是<b>一段</b>文本</p>"
text = re.sub(r"<[^>]+>", "", html) # 这是一段文本
# 2. 提取邮箱
text = "联系：test@example.com 或 admin@company.cn"
emails = re.findall(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", text)
# 3. 清理多余空白
text = " 多 个 空格 "
cleaned = re.sub(r"\s+", " ", text).strip() # 多 个 空格
# 4. 批量替换模式（数据清洗常用）
RE_WELFARE = re.compile(r"(五险一金|带薪年假|节日福利|补充医疗保险)", re.IGNORECASE)
RE_CONTACT = re.compile(r"(联系电话|联系人|Tel|Phone)[：:]\s*\S+", re.IGNORECASE)
def clean_text(text):
text = RE_WELFARE.sub("", text)
text = RE_CONTACT.sub("", text)
return text.strip()
十二、pathlib 模块（现代路径处理）
pathlib 是 Python 3.4+ 推荐的路径处理方式，比 os.path 更优雅。
from pathlib import Path
# 创建路径对象
p = Path("data") / "output" / "result.csv"
print(p) # data/output/result.csv
# 路径属性
p.name # 'result.csv'
p.stem # 'result'
p.suffix # '.csv'
p.parent # Path('data/output')
# 文件操作
p.exists() # 是否存在
p.is_file() # 是否为文件
p.is_dir() # 是否为目录
p.read_text() # 读取文本内容
p.write_text("内容") # 写入文本
# 目录操作
Path("new_dir").mkdir(parents=True, exist_ok=True) # 递归创建目录
# 遍历目录
for f in Path("data").glob("*.csv"):
print(f.name)
# 递归遍历
for f in Path("data").rglob("*.csv"):
print(f)
十三、collections 模块（高级数据结构）
13.1 Counter - 计数器
from collections import Counter
# 统计词频
words = ["苹果", "香蕉", "苹果", "橙子", "香蕉", "苹果"]
counter = Counter(words)
counter.most_common(2) # [('苹果', 3), ('香蕉', 2)]
counter["苹果"] # 3
counter["西瓜"] # 0（不存在返回0）
# 合并计数器
c1 = Counter(a=1, b=2)
c2 = Counter(b=3, c=4)
c1 + c2 # Counter({'b': 5, 'c': 4, 'a': 1})
13.2 defaultdict - 默认字典
from collections import defaultdict
# 自动初始化不存在的键
groups = defaultdict(list)
students = [
("张三", "A班"), ("李四", "B班"), ("王五", "A班")
]
for name, cls in students:
groups[cls].append(name)
# {'A班': ['张三', '王五'], 'B班': ['李四']}
# 统计词频（不用setdefault）
word_count = defaultdict(int)
for word in ["a", "b", "a", "c", "b", "a"]:
word_count[word] += 1
# {'a': 3, 'b': 2, 'c': 1}
13.3 OrderedDict - 有序字典
from collections import OrderedDict
# Python 3.7+ 普通dict已保证插入顺序
# OrderedDict 额外支持 move_to_end() 和相等比较（考虑顺序）
od = OrderedDict()
od["a"] = 1
od["b"] = 2
od["c"] = 3
od.move_to_end("a") # 移到末尾
# OrderedDict([('b', 2), ('c', 3), ('a', 1)])
十四、类型提示（代码质量保障）
类型提示让代码更易读、更易维护，配合 IDE 可以提前发现错误。
14.1 基本类型提示
def greet(name: str) -> str:
return f"Hello, {name}"
def calculate_average(scores: list[float]) -> float:
return sum(scores) / len(scores)
# 变量类型提示
age: int = 25
name: str = "张三"
scores: list[int] = [85, 92, 78]
14.2 复杂类型提示
from typing import Dict, List, Set, Tuple, Optional, Union
# 字典类型
def get_user(name: str) -> Dict[str, Union[str, int]]:
return {"name": name, "age": 25}
# 可选类型（可能为None）
def find_user(user_id: int) -> Optional[Dict[str, str]]:
if user_id > 0:
return {"name": "张三"}
return None
# 元组类型
def get_coordinates() -> Tuple[float, float]:
return (116.4, 39.9)
# 集合类型
def get_unique_tags(text: str) -> Set[str]:
return set(text.split(";"))
14.3 typing 模块高级用法
from typing import Callable, Any
# 回调函数类型
def process_data(data: list, callback: Callable[[Any], Any]) -> list:
return [callback(item) for item in data]
# Python 3.10+ 新语法
def greet(name: str | None = None) -> str:
return f"Hello, {name or 'World'}"
十五、日志系统（生产级数据处理）
15.1 基本使用
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
# 使用
logging.info("开始处理数据...")
logging.warning("发现缺失值")
logging.error("文件读取失败")
15.2 日志级别
| 级别 | 数值 | 用途 |
|---|---|---|
DEBUG | 10 | 调试信息 |
INFO | 20 | 一般信息 |
WARNING | 30 | 警告 |
ERROR | 40 | 错误 |
CRITICAL | 50 | 严重错误 |
15.3 实战配置
import logging
# 创建 logger
logger = logging.getLogger("data_pipeline")
logger.setLevel(logging.DEBUG)
# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 文件处理器
file_handler = logging.FileHandler("pipeline.log", encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
# 格式
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
logger.addHandler(console_handler)
logger.addHandler(file_handler)
# 使用
logger.info("Pipeline 启动")
logger.debug("处理第 1000 条记录")
十六、数据处理进阶技巧
16.1 生成器（节省内存）
# 处理大文件时，逐行读取避免内存溢出
def read_large_file(filepath):
with open(filepath, "r", encoding="utf-8") as f:
for line in f:
yield line.strip()
# 使用
for line in read_large_file("huge_data.csv"):
process(line)
# 生成器表达式（比列表推导式更省内存）
sum(x**2 for x in range(1000000)) # 不会创建完整列表
16.2 上下文管理器
# 自定义上下文管理器
class Timer:
def __enter__(self):
import time
self.start = time.time()
return self
def __exit__(self, *args):
import time
self.elapsed = time.time() - self.start
print(f"耗时: {self.elapsed:.2f}秒")
# 使用
with Timer() as t:
# 执行耗时操作
pass
print(f"总耗时: {t.elapsed}秒")
16.3 dataclass（数据类）
from dataclasses import dataclass
from typing import List
@dataclass
class JobRecord:
"""岗位记录"""
name: str
company: str
salary: str
skills: List[str]
score: float = 0.0
def is_high_salary(self) -> bool:
return "面议" not in self.salary and "k" in self.salary.lower()
# 使用
job = JobRecord("算法工程师", "某科技公司", "30-50k", ["Python", "PyTorch"])
print(job) # JobRecord(name='算法工程师', company='某科技公司', ...)
附录 B：数据分析常用正则模式速查
| 场景 | 正则表达式 | 说明 |
|---|---|---|
| 手机号 | 1[3-9]\d{9} | 11位手机号 |
| 邮箱 | [a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,} | 标准邮箱格式 |
| 身份证 | \d{17}[\dXx] | 18位身份证 |
| IP地址 | \d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3} | IPv4地址 |
| 中文字符 | [一-龥] | 匹配单个中文 |
| HTML标签 | <[^>]+> | 匹配HTML标签 |
| 空白字符 | \s+ | 匹配一个或多个空白 |
| 数字 | \d+ | 匹配一个或多个数字 |
这份笔记涵盖了数据分析方向最常用的 Python 基础知识。建议配合 NumPy、Pandas、Matplotlib 等库一起学习，构建完整的数据分析技能栈。
评论