遍历CSV时如何匹配对应配置表校验列组合唯一性？

我需要遍历S3存储桶中前缀为decrypted/的CSV文件，读取每个文件后，依据unique_columns.json配置文件中对应表的指定列组合，校验该CSV中这些列的组合是否唯一。配置文件定义了各表对应的唯一列组合，比如TABLE_1对应"CLIENT, ADDRNUMBER, PERSNUMBER, CONSNUMBER, LAST_LOAD_DT"。

需求是：读取TABLE_1.csv时校验TABLE_1对应列组合是否唯一，读取TABLE_2.csv时校验TABLE_2对应列组合是否唯一，以此类推。但当前代码存在问题：遍历S3中的CSV文件时，会将每个CSV与配置里的所有表逐一对比，而非按文件名与表名一一对应的方式校验，需要修正该逻辑。

原代码 #

import boto3
import pandas as pd
import json

s3 = boto3.resource('s3')
my_bucket = s3.Bucket('bucket-name-here')

def verify_unique_col():
    with open(r'unique_columns.json', 'r') as f:
        config = json.load(f)

    for my_bucket_object in my_bucket.objects.filter(Prefix='decrypted/'):
        if my_bucket_object.key.endswith('.csv'):
            filename = my_bucket_object.key.split('/')[-1]
            for table in config['Unique_Column_Combination']['Table_Name']:
                unique_col_comb = config['Unique_Column_Combination']['Table_Name'][f'{table}']
                df = pd.read_csv(f's3://path/to/{filename}', sep='|')
                df_unique = df.set_index(unique_col_comb.split(", ")).index.is_unique       
                print(df_unique)

verify_unique_col()

配置文件内容 #

{
    "Unique_Column_Combination":
    {
        "Table_Name":{
            "TABLE_1": "CLIENT, ADDRNUMBER, PERSNUMBER, CONSNUMBER, LAST_LOAD_DT",
            "TABLE_2": "CLIENT, ADDRNUMBER, CONSNUMBER, LAST_LOAD_DT",
            "TABLE_3": "CLIENT, ADDRNUMBER, DATE_FROM, NATION, LAST_LOAD_DT",
            "TABLE_4": "COMM_TYPE, CONSNUMBER, COMM_USAGE, VALID_TO, LOAD_DT"
        }
    }
}

修正后的代码 #

import boto3
import pandas as pd
import json

s3 = boto3.resource('s3')
my_bucket = s3.Bucket('bucket-name-here')

def verify_unique_col():
    with open(r'unique_columns.json', 'r') as f:
        config = json.load(f)
    
    # 提取表名与列组合的映射字典，方便快速匹配
    table_col_map = config['Unique_Column_Combination']['Table_Name']

    for my_bucket_object in my_bucket.objects.filter(Prefix='decrypted/'):
        if my_bucket_object.key.endswith('.csv'):
            filename = my_bucket_object.key.split('/')[-1]
            # 从文件名中提取表名（移除.csv后缀）
            table_name = filename.rsplit('.', 1)[0]
            
            # 仅处理配置中存在的表
            if table_name in table_col_map:
                unique_col_comb = table_col_map[table_name]
                # 构造正确的S3文件路径
                s3_file_path = f's3://{my_bucket.name}/{my_bucket_object.key}'
                df = pd.read_csv(s3_file_path, sep='|')
                # 将列组合字符串转为列表
                unique_cols = unique_col_comb.split(", ")
                # 校验列组合的唯一性
                is_unique = df.set_index(unique_cols).index.is_unique
                print(f"表 {table_name} 的指定列组合是否唯一: {is_unique}")
            else:
                print(f"配置中未找到表 {table_name} 的唯一列组合定义")

verify_unique_col()

关键修改点 #

• 精准匹配表名：从CSV文件名中提取表名（去掉.csv后缀），直接在配置映射中查找对应的列组合，不再遍历所有表
• 优化配置查找：提前将配置中的表名与列组合存入字典，减少嵌套查找的冗余
• 修复S3路径：改用存储桶名称和对象key拼接正确的S3文件路径，避免硬编码路径导致的错误
• 增加异常提示：如果CSV对应的表名不在配置中，输出明确提示信息
• 优化输出可读性：打印结果时标注对应表名，方便快速定位校验结果

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