如何在PySpark中创建15分钟时间桶时间序列，解决员工工时统计数据缺失问题

第一步：优先排查映射函数逻辑错误（核心问题所在） #

你的create_time_block_columns函数存在多处条件判断错误，是数据缺失的主要原因：

• 完全覆盖时段的条件写反：原代码elif (i < inhour) & (i > outhour) 逻辑上永远不可能成立，正确应为i > inhour and i < outhour，即当前小时晚于入职小时、早于离职小时，员工整小时在岗
• 离职时段的条件写反：原代码elif (i < inhour) & (i == outhour) 逻辑错误，正确应为i > inhour and i == outhour，即当前小时是离职小时，且晚于入职小时
• 离职时段的分钟计算逻辑错误：原代码如果outminute < lowerBound 会计算出负数，后续被minutes_worked>0过滤掉，正确判断逻辑应为：如果离岗分钟小于时段下限，该时段计0；如果大于等于时段上限，计15；否则计离岗分钟减时段下限
• Python原生逻辑建议用and替代位运算符&，避免隐式类型转换导致的判断错误

修复后的映射函数参考：

def create_time_block_columns(row_dict):
    inhour = int(row_dict['inhour'])
    outhour = int(row_dict['outhour'])
    inminute = int(row_dict['inminute'])
    outminute = int(row_dict['outminute'])
    # 遍历24小时
    for i in range(24):
        # 遍历每小时4个15分钟时段
        for j in range(1,5):
            lowerBound = (j-1)*15
            upperBound = j*15
            timeBlockColumnName = f't_{i}_{lowerBound}'
            row_dict[timeBlockColumnName] = 0

            # 员工当前时段在岗
            if inhour <= i <= outhour:
                # 入职小时==当前小时，且离职在后续小时
                if i == inhour and outhour > i:
                    if inminute >= upperBound:
                        row_dict[timeBlockColumnName] = 0
                    elif inminute >= lowerBound:
                        row_dict[timeBlockColumnName] = upperBound - inminute
                    else:
                        row_dict[timeBlockColumnName] = 15
                # 当前小时完全在入职和离职小时之间，整时段在岗
                elif inhour < i < outhour:
                    row_dict[timeBlockColumnName] = 15
                # 当前小时是离职小时，且晚于入职小时
                elif i > inhour and i == outhour:
                    if outminute <= lowerBound:
                        row_dict[timeBlockColumnName] = 0
                    elif outminute >= upperBound:
                        row_dict[timeBlockColumnName] = 15
                    else:
                        row_dict[timeBlockColumnName] = outminute - lowerBound
                # 入职和离职都在当前小时
                elif i == inhour and i == outhour:
                    if outminute <= lowerBound or inminute >= upperBound:
                        row_dict[timeBlockColumnName] = 0
                    else:
                        actual_in = max(inminute, lowerBound)
                        actual_out = min(outminute, upperBound)
                        row_dict[timeBlockColumnName] = actual_out - actual_in
    return row_dict

第二步：验证映射结果，排除stack操作问题 #

完成函数修复后，按以下步骤验证：

• 取1~2条已知打卡时间的测试数据，单独调用create_time_block_columns函数，手动核对各时段的分钟计算结果是否符合预期
• 生成mappedDF后，打印单条员工的所有t_开头的列值，确认非零值的时段符合实际打卡时间
• 临时注释掉filter('minutes_worked > 0')代码，统计timeBlockDF的行数，确认行数等于原数据行数乘以96（24*4），如果匹配说明stack操作逻辑正常
• 打印stack_expression字符串，检查是否有列名拼写错误、参数缺失问题

第三步：其他注意事项 #

• 确认源表的inhour、outhour、inminute、outminute字段均为整数类型，避免字符串和数字比较导致的判断失败
• 跨天打卡的场景需要额外处理，比如员工23点入职，次日1点离职，当前逻辑只统计当日时段，会丢失次日数据

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