def _combine_by_key(combine, merge, num_partitions)
        num_partitions ||= default_reduce_partitions

        # Combine key
        combined = new_rdd_from_command(combine.shift, *combine)

        # Merge items
        shuffled = combined.partition_by(num_partitions)
        merge_comm = shuffled.add_command(merge.shift, *merge)

        PipelinedRDD.new(shuffled, merge_comm)
      end
def _combine_by_key(combine, merge, num_partitions)
        num_partitions ||= default_reduce_partitions

        # Combine key
        combined = new_rdd_from_command(combine.shift, *combine)

        # Merge items
        shuffled = combined.partition_by(num_partitions)
        merge_comm = shuffled.add_command(merge.shift, *merge)

        PipelinedRDD.new(shuffled, merge_comm)
      end
For using a different combine_by_key

 == Used for:
 * combine_by_key
 * fold_by_key (with zero value)