本文详细介绍了如何使用pandas库对dataframe中成对出现的源数据（source）和目标数据（target）进行多列匹配，并根据匹配结果在源数据行中添加“pass”或“fail”标记。教程涵盖了数据准备、成对数据对齐、多列比较逻辑实现以及结果整合与列顺序调整等关键步骤，旨在提供一个清晰、高效的解决方案。
问题阐述：DataFrame成对行数据匹配与结果标记
在数据处理和分析中，我们经常会遇到需要比较DataFrame中成对行数据的情况。例如，一个DataFrame可能包含一系列“源数据”（Source）行及其对应的“目标数据”（Target）行。我们的任务是根据这些成对行中特定列的值是否完全匹配，为每一对的“源数据”行添加一个“Result”标记，指示该对数据是“Pass”（匹配）还是“Fail”（不匹配）。
考虑以下示例数据结构：
| Obs | Dataset | Col1 | Col2 | Col3 |
|---|---|---|---|---|
| 1 | Source | A | 10 | X |
| 2 | Target | A | 10 | X |
| 3 | Source | B | 20 | Y |
| 4 | Target | B | 20 | Y |
| 5 | Source | C | 30 | Z |
| 6 | Target | D | 30 | Z |
我们期望的输出结果是：
| Obs | Dataset | Result | Col1 | Col2 | Col3 |
|---|---|---|---|---|---|
| 1 | Source | Pass | A | 10 | X |
| 2 | Target | A | 10 | X | |
| 3 | Source | Pass | B | 20 | Y |
| 4 | Target | B | 20 | Y | |
| 5 | Source | Fail | C | 30 | Z |
| 6 | Target | D | 30 | Z |
注意，Result列只在Dataset为Source的行中显示结果，且其位置在Dataset列之后。
核心策略：基于索引的成对比较
解决这类问题的核心在于如何有效地将“源数据”行与其对应的“目标数据”行进行配对，并执行多列比较。由于示例数据中“Source”和“Target”行是交替出现的，我们可以利用这一点，通过索引操作将它们逻辑上对齐，然后进行向量化比较。
具体步骤如下：
- 数据准备：创建或加载原始DataFrame。
- 分离与对齐：将DataFrame拆分为“Source”行和“Target”行，并通过调整索引将它们按对对齐。
- 执行比较：对齐后的“Source”和“Target”数据进行指定列的逐一比较，判断是否所有列都匹配。
- 标记结果：根据比较结果，在原始DataFrame的“Source”行中更新“Result”列。
- 调整列顺序：将“Result”列移动到期望的位置。
实现步骤
我们将通过python和Pandas库逐步实现上述策略。
步骤一：准备示例数据
首先，我们创建上述示例DataFrame：
import pandas as pd data = { 'Obs': [1, 2, 3, 4, 5, 6], 'Dataset': ['Source', 'Target', 'Source', 'Target', 'Source', 'Target'], 'Col1': ['A', 'A', 'B', 'B', 'C', 'D'], 'Col2': [10, 10, 20, 20, 30, 30], 'Col3': ['X', 'X', 'Y', 'Y', 'Z', 'Z'] } df = pd.DataFrame(data) print("原始DataFrame:") print(df)
步骤二：分离与对齐源数据和目标数据
我们将原始DataFrame分为source_df和target_df。为了将它们对齐，我们利用Obs列的规律（Source行通常是奇数Obs，Target行是偶数Obs，且它们成对出现），通过对原始索引进行整数除法，创建一个逻辑上的“对ID”，然后基于这个“对ID”进行合并。
# 初始化'Result'列 df['Result'] = '' # 分离Source和Target行 source_rows = df[df['Dataset'] == 'Source'].copy() target_rows = df[df['Dataset'] == 'Target'].copy() # 为Source和Target行创建临时的“对ID”，用于对齐 # 假设Source行索引为0, 2, 4...，Target行索引为1, 3, 5... # 那么 (原始索引 // 2) 可以将每对Source/Target映射到同一个ID source_rows['pair_id'] = source_rows.index // 2 target_rows['pair_id'] = target_rows.index // 2 # 基于pair_id合并Source和Target行，以便进行横向比较 # 这里使用 suffixes 来区分合并后的列名 merged_pairs = pd.merge(source_rows, target_rows, on='pair_id', suffixes=('_Source', '_Target')) print("n合并后的成对数据（用于比较）:") print(merged_pairs)
步骤三：执行多列匹配判断
现在merged_pairsDataFrame中，每一行代表一对Source/Target数据。我们可以定义需要比较的列，然后逐一比较这些列，判断是否所有列都匹配。
# 定义需要比较的列 comparison_cols = ['Col1', 'Col2', 'Col3'] # 检查所有比较列是否都匹配 # (merged_pairs[[f'{col}_Source' for col in comparison_cols]].values == ...).all(axis=1) # 这一步将生成一个布尔Series，指示每对数据是否完全匹配 all_cols_match = (merged_pairs[[f'{col}_Source' for col in comparison_cols]].values == merged_pairs[[f'{col}_Target' for col in comparison_cols]].values).all(axis=1) print("n每对数据是否完全匹配（布尔序列）:") print(all_cols_match)
步骤四：标记结果并更新原始DataFrame
根据上一步得到的布尔序列，我们可以生成“Pass”或“Fail”的标记，并将其更新到原始DataFrame的Result列中。由于Result只在Source行显示，我们需要将结果映射回Source行的原始索引。
# 创建一个结果Series，索引对应原始DataFrame中的Source行 # merged_pairs的索引与source_rows的pair_id一致，我们需要将其转换回原始的df索引 source_original_indices = source_rows.index.values # 获取原始Source行的索引 results_series = pd.Series( ['Pass' if match else 'Fail' for match in all_cols_match], index=source_original_indices ) # 更新原始DataFrame的'Result'列 df.loc[results_series.index, 'Result'] = results_series.values print("n更新Result列后的DataFrame:") print(df)
步骤五：调整列顺序以符合输出要求
最后，为了使Result列位于Dataset列之后，我们需要重新排列DataFrame的列。
# 获取当前列的顺序 cols = df.columns.tolist() # 找到'Result'和'Dataset'列的索引 result_col_index = cols.index('Result') dataset_col_index = cols.index('Dataset') # 将'Result'列从当前位置移除 result_col = cols.pop(result_col_index) # 将'Result'列插入到'Dataset'列之后 cols.insert(dataset_col_index + 1, result_col) # 应用新的列顺序 df = df[cols] print("n最终结果DataFrame:") print(df)
完整示例代码
将上述所有步骤整合，得到一个完整的解决方案：
import pandas as pd def determine_matching_pairs(df: pd.DataFrame, comparison_cols: list) -> pd.DataFrame: """ 根据成对的Source/Target行，比较指定列是否匹配，并标记结果。 Args: df (pd.DataFrame): 包含Source和Target行的原始DataFrame。 假定Source和Target行交替出现。 comparison_cols (list): 用于比较的列名列表。 Returns: pd.DataFrame: 包含'Result'列（Pass/Fail）的更新后的DataFrame。 """ # 复制DataFrame以避免修改原始数据 df_processed = df.copy() # 初始化'Result'列 df_processed['Result'] = '' # 分离Source和Target行 source_rows = df_processed[df_processed['Dataset'] == 'Source'].copy() target_rows = df_processed[df_processed['Dataset'] == 'Target'].copy() # 为Source和Target行创建临时的“对ID”，用于对齐 source_rows['pair_id'] = source_rows.index // 2 target_rows['pair_id'] = target_rows.index // 2 # 基于pair_id合并Source和Target行，以便进行横向比较 merged_pairs = pd.merge(source_rows, target_rows, on='pair_id', suffixes=('_Source', '_Target')) # 检查所有比较列是否都匹配 # (merged_pairs[[f'{col}_Source' for col in comparison_cols]].values == ...) # .all(axis=1) 确保所有比较列在同一行都匹配 all_cols_match = (merged_pairs[[f'{col}_Source' for col in comparison_cols]].values == merged_pairs[[f'{col}_Target' for col in comparison_cols]].values).all(axis=1) # 创建一个结果Series，索引对应原始DataFrame中的Source行 source_original_indices = source_rows.index.values results_series = pd.Series( ['Pass' if match