如何从参数文件读取无效prty_id，过滤Spark DataFrame有效Party ID

核心思路 #

先从参数文件读取所有无效prty_id，再从最新Party记录DataFrame中排除这些ID，得到有效数据集。以下是几种常见参数文件格式的具体实现（以Python Pandas为例）：

1. 纯文本格式（每行一个无效ID） #

假设参数文件invalid_prty_ids.txt内容为：

1001
1003
1005

实现代码：

# 读取无效ID列表，去除空行并转换类型
with open('invalid_prty_ids.txt', 'r') as f:
    invalid_ids = [line.strip() for line in f if line.strip()]
# 确保和DataFrame中prty_id类型一致，比如转成int
invalid_ids = list(map(int, invalid_ids))

# 过滤有效记录（latest_party_df为你的最新Party数据集）
valid_party_df = latest_party_df[~latest_party_df['prty_id'].isin(invalid_ids)]

2. CSV格式 #

假设参数文件invalid_prty_ids.csv内容为：

prty_id
1001
1003
1005

实现代码：

import pandas as pd

# 读取CSV文件提取无效ID
invalid_ids_df = pd.read_csv('invalid_prty_ids.csv')
invalid_ids = invalid_ids_df['prty_id'].tolist()

# 过滤有效记录
valid_party_df = latest_party_df[~latest_party_df['prty_id'].isin(invalid_ids)]

3. JSON格式 #

假设参数文件invalid_prty_ids.json内容为：

{"invalid_prty_ids": [1001, 1003, 1005]}

实现代码：

import json

# 读取JSON文件获取无效ID列表
with open('invalid_prty_ids.json', 'r') as f:
    config = json.load(f)
invalid_ids = config['invalid_prty_ids']

# 过滤有效记录
valid_party_df = latest_party_df[~latest_party_df['prty_id'].isin(invalid_ids)]

注意事项 #

• 必须保证读取的无效ID类型和DataFrame中prty_id的类型完全一致（比如都是整数或字符串），否则会出现匹配失效的情况。
• 如果无效ID数据集很大，推荐用merge方式过滤，避免内存占用过高：

valid_party_df = latest_party_df.merge(
    invalid_ids_df, 
    on='prty_id', 
    how='left', 
    indicator=True
).query('_merge == "left_only"').drop('_merge', axis=1)

内容的提问来源于stack exchange，提问作者Silpa Pinjala