再考虑一下reduce侧连接的map reduce管道。看来你很难理解。
为了从两个关系中区分一个键-值对，必须在Map器产生的值上添加一个关系符号。假设你想做一个内部连接，你必须 yield 在reduce中的一个元组，用于reduce-side连接，只有在 contracts 还有你的 transactions 数据集。因此，必须将这些关系的元组保存在单独的列表中，并通过关系符号标识元组。这可以很容易地调整其他连接-例如（左/右/全）外部连接，半/反连接。
在下面的示例中，我使用了关系符号 'C' 对于 contracts 以及 'T' 对于 transactions 数据集。我不能亲自尝试，因为我缺乏数据集，但它应该这样工作。如果你有任何问题，请给我一个评论。
我可以建议您看看“donald miner，adam Shake的mapreduce设计模式”一书，因为它还解释了mapreduce任务的常见连接算法。同时查看最新的mrjob文档。

from mrjob.job import MRJob
from mrjob.step import MRStep
class repartition_join(MRJob):
    def mapper(self, _, line):
        fields=line.split(',')
        if len(fields == 5): # contracts dataset
            join_key = fields[0] # key is in attribute address
            yield (join_key, ('C', 1)) # yield join key, value not used
        elif len(fields) == 7: # transactions dataset
            join_key = fields[2] # key is in attribute to_address
            join_value = int(fields[3]) # value is in attribute value
            yield (join_key, ('T', join_value)) # yields join key with value
        else:
            pass # TODO handle error
    def reducer(self, key, values):
        address = key # the join key
        contracts_tuples = []
        transactions_tuples = []
        for value in values:
            relation_symbol = value[0] # either 'T' or 'C'
            if relation_symbol == 'C': # contracts dataset
                contracts_tuples.append(value[1]) # always 1 - just to know that there is a tuple in contracts
            elif relation_symbol == 'T': # transactions dataset
                transactions_tuples.append(value[1]) # append the value inside value attribute
            else:
                pass # TODO handle error
        # inner join contract and transaction, generalize if needed
        if len(contracts_tuples) > 0 and len(transactions_tuples) > 0:
            for value in transactions_tuples:
                yield (address, value)
    def steps(self):
        return [MRStep(
            mapper=self.mapper,
            reducer=self.reducer)
        ]
if __name__=='__main__':
    repartition_join.run()