基于Pandas实现DataFrame成对行匹配与结果标记

本文详细介绍了如何使用pandas库对dataframe中成对出现的源数据（source）和目标数据（target）进行多列匹配，并根据匹配结果在源数据行中添加“pass”或“fail”标记。教程涵盖了数据准备、成对数据对齐、多列比较逻辑实现以及结果整合与列顺序调整等关键步骤，旨在提供一个清晰、高效的解决方案。

问题阐述：DataFrame成对行数据匹配与结果标记

在数据处理和分析中，我们经常会遇到需要比较DataFrame中成对行数据的情况。例如，一个DataFrame可能包含一系列“源数据”（Source）行及其对应的“目标数据”（Target）行。我们的任务是根据这些成对行中特定列的值是否完全匹配，为每一对的“源数据”行添加一个“Result”标记，指示该对数据是“Pass”（匹配）还是“Fail”（不匹配）。

考虑以下示例数据结构：

我们期望的输出结果是：

注意，Result列只在Dataset为Source的行中显示结果，且其位置在Dataset列之后。

核心策略：基于索引的成对比较

解决这类问题的核心在于如何有效地将“源数据”行与其对应的“目标数据”行进行配对，并执行多列比较。由于示例数据中“Source”和“Target”行是交替出现的，我们可以利用这一点，通过索引操作将它们逻辑上对齐，然后进行向量化比较。

具体步骤如下：

1. 数据准备：创建或加载原始DataFrame。
2. 分离与对齐：将DataFrame拆分为“Source”行和“Target”行，并通过调整索引将它们按对对齐。
3. 执行比较：对齐后的“Source”和“Target”数据进行指定列的逐一比较，判断是否所有列都匹配。
4. 标记结果：根据比较结果，在原始DataFrame的“Source”行中更新“Result”列。
5. 调整列顺序：将“Result”列移动到期望的位置。

实现步骤

我们将通过Python和Pandas库逐步实现上述策略。

阿里云AI平台

阿里云AI平台

下载

步骤一：准备示例数据

首先，我们创建上述示例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