Python中MRJob子类调用super()触发TypeError问题求助

问题场景 #

在Python 3.x环境下，自定义的MRJob子类MapReduceJoinJob在Jupyter Notebook中单元测试正常，但保存为.py文件在控制台运行时触发TypeError。报错发生在MapReduceJoinJob的configure_args方法中，调用super(JoinJob, self).configure_args()时提示：

TypeError: super(type, obj): obj must be an instance or subtype of type

报错核心原因 #

1. super参数匹配错误：MapReduceJoinJob继承的是MRJob，但调用super时传入的第一个参数是JoinJob——而MapReduceJoinJob和JoinJob没有任何继承关系，self是MapReduceJoinJob的实例，自然不属于JoinJob的实例或子类型，触发类型校验错误。
2. JoinJob类的递归调用问题：JoinJob的configure_args方法中直接调用JoinJob.configure_args()，没有传入self，会导致无限递归调用，即使没触发当前报错，后续也会引发栈溢出。
3. 类内代码执行时机错误：在MapReduceJoinJob类定义内部直接实例化JoinJob并调用testJob()，这部分代码会在类加载阶段就自动执行，而非在主程序逻辑中按需执行，打乱了正常的运行流程。

修复步骤 #

1. 修正MapReduceJoinJob的super调用 #

将super(JoinJob, self).configure_args()改为以下两种方式之一（Python 3推荐省略参数的写法）：

# 写法1：Python3自动推导父类
def configure_args(self):
    super().configure_args()

# 写法2：明确指定当前类和self
def configure_args(self):
    super(MapReduceJoinJob, self).configure_args()

2. 修复JoinJob的configure_args方法 #

如果JoinJob的configure_args方法没有实际必要的逻辑，可以直接删除；如果需要保留，要避免递归调用，比如改成：

class JoinJob():
    def configure_args(self):
        # 这里添加实际的配置逻辑，比如初始化参数等
        pass

3. 调整代码执行时机 #

把JoinJob的实例化和方法调用移到if __name__ == "__main__"块中，避免类加载时自动执行：

if __name__ == "__main__":
    driver_reduce_join()
    # 执行JoinJob的测试逻辑
    join_instance = JoinJob()
    join_instance.testJob()
    # 初始化MapReduceJoinJob实例
    mr_instance = MapReduceJoinJob()

完整修正后的代码示例 #

# MRJob MapReduce using MRJob (MRUnit in Python)
# MRJob / MRUnit Test unit class

import pandas as pd
from mrjob.job import MRJob
from mrjob.protocol import RawValueProtocol

# 假设以下函数是已实现的：MapReduceJoin、CombineDatasets、SortVictimData、driver_reduce_join
def MapReduceJoin():
    # 示例返回数据
    return [], []

def CombineDatasets(h_data, v_data):
    return []

def SortVictimData(df, sort_order=False, column=''):
    return df

def driver_reduce_join():
    pass

class JoinJob():
    # 修复：移除递归调用，添加实际逻辑或留空
    def configure_args(self):
        pass

    def showResults(self, df_h_data, df_v_data, combine_row):
        print('MRJob Test Case : Columns from Dataset Homocides Non-Fatal')
        print('\n')
        print(df_h_data)

        print('MRJob Test Case : Columns from Dataset Victim Demographics')
        print('\n')
        print(df_v_data)

        print('MRJob Test Case : Combined Map Reduced Dataset Homocides Non-Fatal & Victim Demographics')
        print('\n')
        print(combine_row)

    # test class function
    def testJob(self):
        h_data, v_data = MapReduceJoin()
        combined_row = CombineDatasets(h_data, v_data)
        df_combine_row = pd.DataFrame(combined_row)
        by='BATTERY'
        combine_row = SortVictimData(df_combine_row, sort_order=False, column=by)
        df_h_data = pd.DataFrame(h_data)
        df_v_data = pd.DataFrame(v_data)

        self.showResults(df_h_data, df_v_data, combine_row)

# MRJob MapReduce using MRJob (MRUnit in Python)
# Test: test different row filter, test sort

class MapReduceJoinJob(MRJob):
    OUTPUT_PROTOCOL = RawValueProtocol

    def configure_args(self):
        # 修复：调用MRJob的configure_args方法
        super().configure_args()

if __name__ == "__main__":
    driver_reduce_join()
    # 执行JoinJob测试
    join_instance = JoinJob()
    join_instance.testJob()
    # 初始化MapReduceJoinJob实例
    mr_instance = MapReduceJoinJob()
    # 如果需要运行MRJob，调用run方法
    # mr_instance.run()

内容的提问来源于stack exchange，提问作者Data Science Analytics Manager