大规模数据清洗 Pipeline 技术介绍
基于 1600 万行岗位数据清洗项目的实战经验总结
一、项目概述
1.1 项目背景
在 AI 人才分析场景中，需要从各大招聘平台抓取海量岗位数据，但原始数据存在大量噪声：岗位名称不规范、职责描述包含广告、重复数据泛滥、非 AI 岗位混入等问题。本项目设计了一套 6 阶段流水线，将 1600 万行原始数据清洗为高质量的 AI 岗位池。
1.2 核心挑战
| 挑战 | 具体表现 |
|---|---|
| 数据量大 | 1600 万行，单机处理需优化内存和速度 |
| 噪声多 | 包含福利、联系方式、公司介绍等无关文本 |
| 分类模糊 | "运营"、"测试"等岗位需结合上下文判断 |
| 重复泛滥 | 同一公司同一岗位多次发布，内容高度相似 |
| 语义鸿沟 | 传统关键词匹配无法理解岗位语义 |
1.3 整体架构
原始数据 (1600w)
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Stage 1: 文本清洗 (first_cleaned.py) │
│ - 正则表达式清洗噪声文本 │
│ - 提取技能/领域/动作/场景/角色标签 │
│ - 初步分类：accept_direct / reject_direct / to_embedding │
└─────────────────────────────────────────────────────────────┘
│
├──────────────────┬──────────────────┐
▼ ▼ ▼
岗位字典匹配 模糊岗位 直接拒绝
(精筛) (需进一步判断)
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ Stage 2: 字典精筛 │ │ Stage 3: 技能筛选 │
│ (second.py) │ │ (third.py) │
│ - 加权评分 │ │ - 技能标签匹配 │
│ - 去重 │ │ - 英文假阳性过滤 │
│ - 后置规则覆盖 │ │ - 全局去重 │
└─────────────────┘ └─────────────────┘
│ │
▼ ▼
┌─────────────────────────────────────────────────────────────┐
│ Stage 4: 语义聚类 (fourth.py) │
│ - SentenceTransformer 向量化 │
│ - UMAP 降维 │
│ - HDBSCAN 密度聚类 │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Stage 5: 池合并与技能恢复 (fifth.py) │
│ - 合并字典池和模糊池 │
│ - 从 reject 中恢复有技能标签的记录 │
│ - 最终去重 │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Stage 6: AI 行业特征富化 (final.py) │
│ - 三通道评分：标签优先 / 文本规则 / 兜底 │
│ - AI 专业映射 │
│ - 广东省重点产业赛道匹配 │
└─────────────────────────────────────────────────────────────┘
│
▼
最终输出 (高质量 AI 岗位池)
二、Stage 1：文本清洗与标签提取
文件：first_cleaned.py
2.1 核心思路
原始 JD（Job Description）文本充满噪声，第一步是用正则表达式清洗无关内容，同时提取结构化标签。
2.2 噪声清洗规则
# 各类噪声的正则模式
RE_WELFARE = re.compile(
r"(五险一金|带薪年假|节日福利|补充医疗保险|交通补贴|餐饮补贴|住房补贴|"
r"定期体检|员工旅游|年终奖金|绩效奖金|股票期权|弹性工作|免费班车|免费工作餐)"
)
RE_CONTACT = re.compile(
r"(联系人|联系电话|联系方式|Tel|Phone|邮箱|Email|地址|工作地点)[：:]\s*\S+"
)
RE_WORKTIME = re.compile(
r"(工作时间|上班时间|工作制度)[：:]\s*\S+"
)
RE_SOFTSKILL = re.compile(
r"(沟通能力|团队合作|责任心强|抗压能力|学习能力|执行力|积极主动|"
r"良好的沟通|较强的责任心|吃苦耐劳)"
)
RE_COMPANY_DESC = re.compile(
r"(公司简介|公司介绍|关于我们|企业介绍|公司成立于|公司规模|注册资本)"
)
RE_GARBAGE_SECTION = re.compile(
r"(岗位职责|任职要求|职位描述|职位要求|岗位要求|工作职责|岗位信息|"
r"职位信息|薪资福利|福利待遇|晋升空间|发展空间|培训机会)"
)
2.3 标签提取系统
项目定义了 5 个维度的标签提取：
# 1. 技能标签（30+ 技术）
SKILL_PATTERNS = {
"Python": re.compile(r"\bpython\b", re.I),
"Java": re.compile(r"\bjava\b", re.I),
"PyTorch": re.compile(r"\bpytorch\b", re.I),
"TensorFlow": re.compile(r"\btensorflow\b", re.I),
"OpenCV": re.compile(r"\bopencv\b", re.I),
"Transformer": re.compile(r"\btransformer\b", re.I),
"BERT": re.compile(r"\bbert\b", re.I),
"GPT": re.compile(r"\bgpt\b", re.I),
# ... 更多技能
}
# 2. 领域标签
DOMAIN_PATTERNS = {
"自然语言处理": re.compile(r"(NLP|自然语言处理|文本处理|语义理解)"),
"计算机视觉": re.compile(r"(CV|计算机视觉|图像识别|目标检测)"),
"大模型": re.compile(r"(大模型|LLM|大语言模型|GPT)"),
"机器学习": re.compile(r"(机器学习|深度学习|神经网络)"),
"自动驾驶": re.compile(r"(自动驾驶|无人驾驶|ADAS)"),
# ... 更多领域
}
# 3. 动作标签
ACTION_PATTERNS = {
"训练": re.compile(r"(训练|train|training)"),
"部署": re.compile(r"(部署|deploy|deployment)"),
"优化": re.compile(r"(优化|optimize|optimization)"),
"评测": re.compile(r"(评测|评估|benchmark)"),
}
# 4. 场景标签
SCENE_PATTERNS = {
"搜索": re.compile(r"(搜索|search|query)"),
"推荐": re.compile(r"(推荐|recommend)"),
"风控": re.compile(r"(风控|反欺诈|风险控制)"),
}
# 5. 角色标签
ROLE_PATTERNS = {
"算法": re.compile(r"(算法|algorithm)"),
"标注": re.compile(r"(标注|label|annotation)"),
"测试": re.compile(r"(测试|test|testing)"),
}
2.4 标签置信度评分
def compute_tag_confidence(record):
"""根据标签数量和质量计算置信度"""
skill_count = len(split_tags(record.get("jd_skill_tags", "")))
domain_count = len(split_tags(record.get("jd_domain_tags", "")))
if skill_count >= 3 and domain_count >= 2:
return "high"
elif skill_count >= 2 or domain_count >= 2:
return "medium"
elif skill_count >= 1 or domain_count >= 1:
return "low"
else:
return "very_low"
2.5 路由决策
def make_routing_decision(record):
"""决定记录去向"""
bucket = classify_job_title(record.get("name", ""))
confidence = record.get("tag_confidence", "very_low")
# 直接接受：正向岗位 + 高置信度
if bucket == "positive" and confidence in ("high", "medium"):
return "accept_direct"
# 直接拒绝：负向岗位
if bucket == "negative":
return "reject_direct"
# 其他进入 embedding 阶段
return "to_embedding"
三、Stage 2：岗位字典精筛
文件：second.py
3.1 加权评分系统
对于通过字典匹配的岗位，使用加权评分判断是否保留：
# 权重定义
STRONG_DOMAIN_HIGH = {
"自然语言处理": 3.2,
"计算机视觉": 3.2,
"大模型": 3.0,
"机器学习": 3.0,
"自动驾驶": 3.0,
}
MEDIUM_SKILL_WEIGHTS = {
"Python": 1.5,
"Java": 1.2,
"C++": 1.2,
"PyTorch": 2.0,
"TensorFlow": 2.0,
"OpenCV": 1.8,
"Transformer": 1.8,
}
# 评分逻辑
def calculate_score(record):
score = 0.0
# 领域加分
domain_tags = split_tags(record.get("jd_domain_tags", ""))
for tag in domain_tags:
score += STRONG_DOMAIN_HIGH.get(tag, 0)
# 技能加分
skill_tags = split_tags(record.get("jd_skill_tags", ""))
for tag in skill_tags:
score += MEDIUM_SKILL_WEIGHTS.get(tag, 0)
# 动作加分
action_tags = split_tags(record.get("jd_action_tags", ""))
if "训练" in action_tags:
score += 2.0
if "部署" in action_tags:
score += 1.5
return score
3.2 去重策略
def build_duplicate_signature(record):
"""构建去重签名：公司 + 关键词 + 职责"""
company = normalize_text_key(record.get("company", ""))
keyword = normalize_text_key(record.get("keyword", ""))
responsibility = normalize_text_key(record.get("responsibility_compact", ""))
base = f"{company}|{keyword}|{responsibility}"
return hashlib.md5(base.encode("utf-8")).hexdigest()
3.3 后置规则覆盖
某些特殊情况需要覆盖评分结果：
def apply_post_rules_override(record):
"""后置规则：强制拒绝某些情况"""
name = str(record.get("name", ""))
# 硬负向标题
if contains_any_term(name, HARD_NEGATIVE_TITLE_TERMS):
record["final_decision"] = "reject"
record["final_reason"] = "hard_negative_title_override"
return record
# 运营类岗位（除非有强AI特征）
if should_reject_operation_case(record):
record["final_decision"] = "reject"
record["final_reason"] = "operation_title_override"
return record
# 弱搜索案例（只有搜索标签，无NLP/LLM技能）
if is_weak_search_case(record):
record["final_decision"] = "reject"
record["final_reason"] = "weak_search_case_override"
return record
return record
四、Stage 3：模糊岗位技能筛选
文件：third.py
4.1 技能直留规则
对于无法通过字典匹配的"模糊"岗位，使用技能标签判断：
# 强技能标签（命中1个即保留）
STRONG_SKILL_TAGS = {
"PyTorch", "TensorFlow", "OpenCV", "PCL", "CUDA",
"Transformer", "BERT", "GPT", "LLaMA", "LangChain",
"HuggingFace", "RAG", "LoRA", "CNN", "RNN", "GAN",
"XGBoost", "LightGBM", "Scikit-learn",
}
# 中等技能标签（命中2个即保留）
MEDIUM_SKILL_TAGS = {
"Python", "Java", "C++", "C#", "Go", "MATLAB",
"SQL", "MySQL", "PostgreSQL", "MongoDB", "Redis",
"ElasticSearch", "Linux", "Shell", "Docker", "Kubernetes",
"Spark", "Hadoop", "Hive", "Pandas", "NumPy", "Matplotlib",
}
def should_direct_keep_by_skill(record):
"""技能直留判断"""
name = str(record.get("name", ""))
# 硬负向标题直接拒绝
if contains_any_term(name, HARD_NEGATIVE_TITLE_TERMS):
return False
# 英文假阳性过滤
if is_english_false_positive(record):
return False
# 强技能命中
if has_strong_skill(record):
return True
# 中等技能命中2个
if has_medium_skill_pair(record):
return True
return False
4.2 英文假阳性检测
某些岗位虽然包含技术关键词，但实际是纯英文环境的非技术岗：
def is_english_false_positive(record):
"""检测英文假阳性"""
resp = str(record.get("responsibility_compact", ""))
req = str(record.get("requirement_compact", ""))
merged = f"{resp} {req}".strip()
ratio = english_ratio(merged) # 英文字符占比
has_domain = has_core_ai_domain(record)
# 英文占比过高且无AI领域标签 → 假阳性
if ratio >= 0.55 and not has_domain:
return True
# 英文占比过高且技能标签弱 → 假阳性
if ratio >= 0.55 and strong_skill_count == 0 and medium_skill_count <= 2:
return True
return False
4.3 全局去重
所有保留的记录进入候选池，按时间保留最新：
# 候选池：签名 -> (时间, 记录)
candidate_pool = {}
for row in accepted_rows:
sig = build_duplicate_signature(row)
time_val = str(row.get("time", "") or "")
if sig not in candidate_pool or time_val > candidate_pool[sig][0]:
candidate_pool[sig] = (time_val, row)
五、Stage 4：语义向量聚类
文件：fourth.py
5.1 向量化
使用 SentenceTransformer 将岗位文本转换为向量：
from sentence_transformers import SentenceTransformer
# 加载中文 BERT 模型
MODEL_NAME = "D:/models/bge-large-zh-v1.5"
model = SentenceTransformer(MODEL_NAME, device="cuda")
def build_embedding_text(record):
"""构建用于向量化的文本"""
parts = []
if record.get("name"):
parts.append(f"岗位名称: {record['name']}")
if record.get("keyword"):
parts.append(f"关键词: {record['keyword']}")
if record.get("jd_domain_tags"):
parts.append(f"领域标签: {record['jd_domain_tags']}")
if record.get("jd_skill_tags"):
parts.append(f"技能标签: {record['jd_skill_tags']}")
if record.get("responsibility_compact"):
parts.append(f"职责: {record['responsibility_compact']}")
if record.get("requirement_compact"):
parts.append(f"要求: {record['requirement_compact']}")
return "\n".join(parts)
# 批量编码
embeddings = model.encode(
texts,
batch_size=256,
normalize_embeddings=True,
show_progress_bar=True
)
5.2 原型分类
定义 8 个类别原型，用余弦相似度分类：
CATEGORY_PROTOTYPES = {
"nlp_llm": "自然语言处理、大语言模型、LLM、文本生成、信息抽取...",
"cv_vision": "计算机视觉、CV、图像识别、目标检测、图像分割...",
"autonomous_robotics": "自动驾驶、无人驾驶、ADAS、SLAM、机器人...",
"search_recommend_risk": "搜索算法、推荐系统、风控、反欺诈...",
"data_labeling_training": "数据标注、模型评测、数据清洗、AI训练师...",
"mlops_deployment_eval": "模型部署、推理服务、MLOps、性能优化...",
"ai_product_solution": "AI产品经理、智能体、Agent、解决方案...",
"other_ai": "人工智能、AI、机器学习、深度学习、算法研发...",
}
def classify_one(embedding, prototype_embeddings):
"""基于余弦相似度分类"""
scores = []
for category, proto_vec in prototype_embeddings.items():
sim = cosine_similarity(embedding, proto_vec)
scores.append((category, sim))
scores.sort(key=lambda x: x[1], reverse=True)
top1_cat, top1_score = scores[0]
top2_cat, top2_score = scores[1]
return {
"embedding_category_top1": top1_cat,
"embedding_category_score_top1": f"{top1_score:.4f}",
"embedding_category_confidence": judge_confidence(top1_score, top2_score),
}
def judge_confidence(top1_score, top2_score):
"""判断分类置信度"""
if top1_score >= 0.75 and (top1_score - top2_score) >= 0.08:
return "high"
if top1_score >= 0.65 and (top1_score - top2_score) >= 0.03:
return "medium"
return "low"
5.3 UMAP 降维
将高维向量降到低维，便于聚类：
import umap
umap_reducer = umap.UMAP(
n_neighbors=15, # 邻居数量
n_components=10, # 降维后的维度
min_dist=0.0, # 最小距离（0表示允许紧密聚类）
metric='cosine', # 使用余弦距离
random_state=42
)
reduced = umap_reducer.fit_transform(vectors)
参数说明：
n_neighbors=15：控制局部与全局结构的平衡，值越大越关注全局n_components=10：降到 10 维，保留主要结构信息min_dist=0.0：允许点紧密聚集，有利于发现密集簇
5.4 HDBSCAN 密度聚类
import hdbscan
clusterer = hdbscan.HDBSCAN(
min_cluster_size=10, # 最小簇大小
min_samples=5, # 核心点最小邻居数
metric='euclidean', # 距离度量
cluster_selection_method='eom' # 簇选择方法
)
labels = clusterer.fit_predict(reduced)
# 结果解读
# -1 表示噪声点（不属于任何簇）
# 0, 1, 2... 表示簇编号
参数说明：
min_cluster_size=10：簇至少包含 10 个样本，小于此数的被视为噪声min_samples=5：核心点需要至少 5 个邻居，控制簇的密度cluster_selection_method='eom'：Excess of Mass 方法，倾向于选择更紧凑的簇
5.5 类内聚类
对每个类别分别进行聚类，发现子类：
# 按类别分组
cat_to_indices = {}
for idx, row in enumerate(rows):
cat = row["embedding_category_top1"]
cat_to_indices.setdefault(cat, []).append(idx)
# 对每个类别单独聚类
for cat, indices in cat_to_indices.items():
if len(indices) < HDBSCAN_MIN_CLUSTER_SIZE * 2:
# 样本太少，跳过聚类
for idx in indices:
rows[idx]["sub_cluster_id"] = f"{cat}_0"
continue
# 降维
cat_vectors = vectors_array[indices]
reduced = umap_reducer.fit_transform(cat_vectors)
# 聚类
labels = hdbscan_clusterer.fit_predict(reduced)
# 标记结果
for j, idx in enumerate(indices):
if labels[j] == -1:
rows[idx]["sub_cluster_id"] = f"{cat}_noise"
else:
rows[idx]["sub_cluster_id"] = f"{cat}_{labels[j]}"
六、Stage 5：池合并与技能恢复
文件：fifth.py
6.1 池合并
将多个来源的岗位合并：
# 四个来源
dict_final_rows = read_csv_rows("岗位字典_final.csv")
fuzzy_final_rows = read_csv_rows("模糊岗位_final.csv")
dict_recovered_rows = read_csv_rows("岗位字典_reject_skill_recovered.csv")
fuzzy_recovered_rows = read_csv_rows("模糊岗位_reject_skill_recovered.csv")
# 标记来源
for row in dict_final_rows:
row["pool_source"] = "岗位字典_final"
for row in fuzzy_final_rows:
row["pool_source"] = "模糊岗位_final"
# 合并
final_pool = dict_final_rows + fuzzy_final_rows + dict_recovered_rows + fuzzy_recovered_rows
6.2 技能恢复
从 reject 池中恢复有技能标签的记录：
def recover_skill_rows(rows, source_name):
"""从 reject 中恢复有技能的记录"""
recovered = []
for row in rows:
strong_skill_count = safe_int(row.get("strong_skill_count", 0))
medium_skill_count = safe_int(row.get("medium_skill_count", 0))
# 强技能 >= 1 或 中等技能 >= 2
if strong_skill_count >= 1 or medium_skill_count >= 2:
# 英文假阳性过滤
if is_english_false_positive(row):
continue
row["recovered_from_reject"] = 1
row["recovered_source"] = source_name
row["recovered_reason"] = (
"strong_skill_recovered"
if strong_skill_count >= 1
else "two_medium_skills_recovered"
)
recovered.append(row)
return recovered
6.3 最终去重
def deduplicate_keep_latest(rows):
"""去重，保留最新记录"""
rows_sorted = sorted(rows, key=lambda r: str(r.get("time", "")), reverse=True)
seen = set()
deduped = []
for row in rows_sorted:
sig = build_duplicate_signature(row)
if sig not in seen:
seen.add(sig)
deduped.append(row)
return deduped
七、Stage 6：AI 行业特征富化
文件：final.py
7.1 三通道评分
def compute_ai_relevance_score(record):
"""三通道评分"""
# Channel A: 标签优先
domain_tags = split_tags(record.get("jd_domain_tags", ""))
skill_tags = split_tags(record.get("jd_skill_tags", ""))
strong_ai_domains = {"自然语言处理", "计算机视觉", "大模型", "机器学习", "自动驾驶"}
strong_ai_skills = {"PyTorch", "TensorFlow", "OpenCV", "Transformer", "BERT", "GPT", "LLaMA"}
domain_hit = len(domain_tags & strong_ai_domains)
skill_hit = len(skill_tags & strong_ai_skills)
if domain_hit >= 2:
return 0.95, "high"
if domain_hit >= 1 and skill_hit >= 1:
return 0.90, "high"
if skill_hit >= 2:
return 0.85, "high"
# Channel B: 文本规则
text = f"{record.get('responsibility_compact', '')} {record.get('requirement_compact', '')}"
ai_keywords = ["人工智能", "深度学习", "神经网络", "算法", "模型训练"]
keyword_hits = sum(1 for kw in ai_keywords if kw in text)
if keyword_hits >= 3:
return 0.80, "medium"
if keyword_hits >= 2:
return 0.70, "medium"
# Channel C: 兜底
return 0.50, "low"
7.2 AI 专业映射
AI_MAJOR_MAP = {
"计算机科学与技术": {
"segments": ["nlp_llm", "cv_vision", "mlops_deployment_eval"],
"clusters": ["算法工程师", "深度学习工程师"],
"courses": ["机器学习", "深度学习", "计算机视觉"],
},
"人工智能": {
"segments": ["nlp_llm", "cv_vision", "autonomous_robotics"],
"clusters": ["AI算法工程师", "机器学习工程师"],
"courses": ["自然语言处理", "计算机视觉", "强化学习"],
},
"数据科学与大数据技术": {
"segments": ["data_labeling_training", "search_recommend_risk"],
"clusters": ["数据分析师", "数据工程师"],
"courses": ["数据挖掘", "大数据处理", "统计学习"],
},
# ... 更多专业
}
7.3 广东省重点产业赛道
GD_AI_INDUSTRY_TRACK_MAP = {
"智能机器人": {
"priority": "P0",
"keywords": ["机器人", "ROS", "机械臂", "运动控制"],
},
"自动驾驶": {
"priority": "P0",
"keywords": ["自动驾驶", "ADAS", "SLAM", "路径规划"],
},
"智能语音": {
"priority": "P1",
"keywords": ["语音识别", "语音合成", "ASR", "TTS"],
},
"计算机视觉": {
"priority": "P0",
"keywords": ["图像识别", "目标检测", "OCR", "人脸识别"],
},
"自然语言处理": {
"priority": "P0",
"keywords": ["NLP", "文本分析", "机器翻译", "情感分析"],
},
# ... 更多赛道
}
八、关键技术总结
8.1 正则表达式
- 预编译：大量匹配时使用
re.compile()提升性能 - IGNORECASE：使用
re.IGNORECASE忽略大小写 - 非捕获分组：使用
(?:...)避免不必要的捕获 - 批量替换：将多个正则模式编译后依次替换
8.2 加权评分
- 分层权重：强领域 > 中等技能 > 弱信号
- 阈值判断：根据总分决定保留/拒绝
- 后置覆盖：特殊规则可以覆盖评分结果
8.3 去重策略
- 签名构建：公司 + 关键词 + 职责的 MD5 哈希
- 时间排序：相同签名保留最新记录
- 两阶段去重：阶段内去重 + 全局去重
8.4 向量化与聚类
- SentenceTransformer：将文本转换为高维向量
- UMAP：降维保留局部和全局结构
- HDBSCAN：密度聚类，自动发现簇数量，识别噪声点
- 类内聚类：对每个类别单独聚类，发现子类
8.5 工程技巧
- MultiCSVWriter：延迟写入表头，支持多输出文件
- 进度打印：每 50000 行打印进度
- 内存优化：使用生成器和流式处理
- 编码兼容：统一使用
utf-8-sig编码
九、数据流转图
┌──────────────────────────────────────────────────────────────────┐
│ 原始数据 (1600w) │
└──────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────┐
│ Stage 1: 文本清洗与标签提取 │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ accept_direct│ │reject_direct│ │to_embedding │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
└─────────┼────────────────┼────────────────┼──────────────────────┘
│ │ │
│ ▼ │
│ 直接拒绝 (噪声) │
│ │
▼ ▼
┌─────────────────────┐ ┌─────────────────────┐
│ Stage 2: 字典精筛 │ │ Stage 3: 技能筛选 │
│ ┌─────────────────┐│ │ ┌─────────────────┐│
│ │岗位字典_final ││ │ │模糊岗位_final ││
│ │岗位字典_reject ││ │ │模糊岗位_reject ││
│ └────────┬────────┘│ │ └────────┬────────┘│
└───────────┼─────────┘ └───────────┼─────────┘
│ │
▼ ▼
┌──────────────────────────────────────────────────────────────────┐
│ Stage 4: 语义向量聚类 │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ SentenceTrans│ │ UMAP │ │ HDBSCAN │ │
│ │ former │ │ 降维 │ │ 密度聚类 │ │
│ └──────────────┘ └─────────────┘ └─────────────┘ │
└──────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────┐
│ Stage 5: 池合并与技能恢复 │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 岗位字典_final + 模糊岗位_final + reject_skill_recovered │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 全局去重 (MD5签名) │
└──────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────┐
│ Stage 6: AI 行业特征富化 │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ AI相关度评分 │ │ 专业映射 │ │ 产业赛道匹配 │ │
│ └──────────────┘ └─────────────┘ └─────────────┘ │
└──────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────┐
│ 最终输出 (高质量AI岗位池) │
│ - embedding_category_top1: 岗位类别 │
│ - sub_cluster_id: 子类编号 │
│ - ai_relevance_score: AI相关度评分 │
│ - ai_mapped_majors: 匹配专业 │
│ - ai_gd_key_industry_track: 产业赛道 │
└──────────────────────────────────────────────────────────────────┘
十、性能优化建议
10.1 内存优化
# 1. 使用生成器替代列表
def process_rows(filepath):
with open(filepath, "r", encoding="utf-8-sig") as f:
for row in csv.DictReader(f):
yield clean_row(row)
# 2. 分批处理
BATCH_SIZE = 10000
batch = []
for row in process_rows("huge_file.csv"):
batch.append(row)
if len(batch) >= BATCH_SIZE:
process_batch(batch)
batch.clear()
# 3. 及时释放内存
import gc
del large_list
gc.collect()
10.2 计算优化
# 1. 正则预编译
PATTERN = re.compile(r"complex_pattern") # 编译一次
PATTERN.search(text) # 多次使用
# 2. 集合查找替代列表查找
VALID_TAGS = {"Python", "Java", "C++"} # 集合 O(1)
tag in VALID_TAGS # 快速判断
# 3. 批量向量化
embeddings = model.encode(texts, batch_size=256) # 批量处理
10.3 I/O 优化
# 1. 使用 utf-8-sig 编码（兼容 Excel）
with open("output.csv", "w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
# 2. 延迟写入表头
class MultiCSVWriter:
def write_row(self, row):
if not self._header_written:
self.fieldnames = list(row.keys())
self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames)
self.writer.writeheader()
self._header_written = True
self.writer.writerow(row)
附录：核心数据结构
岗位记录字段
| 字段名 | 说明 | 示例 |
|---|---|---|
name | 岗位名称 | "算法工程师" |
company | 公司名称 | "某科技公司" |
keyword | 搜索关键词 | "NLP" |
responsibility_compact | 职责描述（清洗后） | "负责NLP模型训练..." |
requirement_compact | 要求描述（清洗后） | "熟悉PyTorch..." |
jd_skill_tags | 技能标签 | "Python;PyTorch;Transformer" |
jd_domain_tags | 领域标签 | "自然语言处理;大模型" |
jd_action_tags | 动作标签 | "训练;部署" |
jd_scene_tags | 场景标签 | "搜索;推荐" |
jd_role_tags | 角色标签 | "算法" |
job_title_bucket | 岗位分类 | "positive" / "boundary" / "negative" |
final_decision | 最终决策 | "accept_direct" / "reject" |
final_reason | 决策原因 | "skill_direct_keep" |
embedding_category_top1 | 向量分类结果 | "nlp_llm" |
sub_cluster_id | 子类编号 | "nlp_llm_3" |
ai_relevance_score | AI相关度评分 | 0.85 |
ai_relevance_level | AI相关度等级 | "high" |
这套 Pipeline 的核心设计思想是：分层过滤 + 多信号融合 + 语义增强。通过规则层快速过滤明显噪声，通过技能标签层保留有价值信号，通过语义向量层发现隐含关联，最终实现从 1600 万行原始数据中精准提取高质量 AI 岗位。
评论