如何在DataFrame列表型字段上映射函数以获取节点最低活动年级

问题描述 #

我已花费数小时排查该问题，十分感谢您的帮助！每个节点可包含多个活动，每个活动最多关联两个在SQL中为ARRAY类型的年级，我的目标是获取每个节点的最低活动年级。我使用ARRAY_TO_STRING(ACTIVITY_GRADE) AS ACTIVITY_GRADE导入ACTIVITY_GRADE字段，不确定是否有更优方式将其导入为可迭代列表？期望得到目标列min_node_grade。

数据示例 #

现有实现代码 #

# split it into two columns
df[['activity_grade_a', 'activity_grade_b']] = df.ACTIVITY_GRADE.str.split(",", expand = True)

# map it to integers so can take min to identify grade
grade_to_index = {"Preschool": -2, "Pre-K": -1, "Kindergarten": 0, "1st Grade": 1, '2nd Grade':2,'3rd Grade':3,'4th Grade':4,'5th Grade':5}

# map to invert the dictionary in order to get it back to text form
inv_map = {v: k for k, v in grade_to_index.items()}

# create columns with the index for the one or two grades. 
df['activity_grade_a_index']=df['activity_grade_a'].replace(grade_to_index)
df['activity_grade_b_index']=df['activity_grade_b'].replace(grade_to_index)

# get minimum of each row across the two columns; axis=1 says looks across columns
df['activity_min_grade_index'] = df[['activity_grade_a_index', 'activity_grade_b_index']].min(axis=1)

# group by node and get the minimum of activity-level minimums, map it to a new field
df['min_node_grade_index']=df.groupby('NODE_NAME')['activity_min_grade_index'].transform('min')

# get the grade back
df['min_node_grade']=df['min_node_grade_index'].replace(inv_map)

优化方案 #

1. SQL数组导入优化 #

如果使用pandas读取SQL，无需用ARRAY_TO_STRING转换。以PostgreSQL为例，直接查询原ARRAY类型字段，pd.read_sql会自动将其解析为Python列表列，后续处理更高效：

SELECT STUDENT_ID, RECORD_ID, NODE_NAME, ACTIVITY_NAME, ACTIVITY_GRADE FROM your_table;

导入后df['ACTIVITY_GRADE']的每个值都是列表（如['PreK', 'Kindergarten']），无需再拆分字符串。

2. 简化计算逻辑 #

通过自定义函数统一处理字符串/列表类型的年级数据，无需拆分多列，逻辑更通用：

import pandas as pd

# 完善年级映射，包含数据中的"PreK"别名
grade_to_index = {
    "Preschool": -2,
    "Pre-K": -1,
    "PreK": -1,
    "Kindergarten": 0,
    "1st Grade": 1,
    "2nd Grade": 2,
    "3rd Grade": 3,
    "4th Grade": 4,
    "5th Grade": 5
}
inv_map = {v: k for k, v in grade_to_index.items()}

def get_activity_min_idx(grade_data):
    # 兼容列表或逗号分隔字符串
    if isinstance(grade_data, list):
        grades = [g.strip() for g in grade_data]
    else:
        grades = [g.strip() for g in grade_data.split(",")]
    # 映射为索引并取最小值
    valid_indices = [grade_to_index[g] for g in grades if g in grade_to_index]
    return min(valid_indices) if valid_indices else None

# 计算每个活动的最低年级索引
df['activity_min_idx'] = df['ACTIVITY_GRADE'].apply(get_activity_min_idx)

# 按节点分组，获取节点级最低年级索引
df['min_node_grade_idx'] = df.groupby('NODE_NAME')['activity_min_idx'].transform('min')

# 映射回年级名称
df['min_node_grade'] = df['min_node_grade_idx'].map(inv_map)

优化优势 #

• 无需拆分多列，代码更紧凑，兼容1个或多个年级的场景（后续活动年级数量变化也无需修改代码）
• 同时支持列表和字符串类型的年级数据，适配不同的导入方式
• 逻辑清晰，可读性更强，减少中间临时列的创建

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