0

mapreduce を使用した疎行列の乗算が可能になるように、高度なチュートリアルの innerjoin-example を変更したいと考えています (Ullman による説明)。したがって、結果マトリックス内の等しい位置の値を合計する 2 番目の map-reduce ステップが必要です。

残念ながら、クラス CsvInnerJoin の最初の reduce 関数の出力を SumJob の map 関数に入れることができません。

import sys
sys.path.append("/home/damian/disco/lib/")
from disco.core import Job, result_iterator
from disco.worker.classic.func import chain_reader
import csv, sys


if __name__ == '__main__':
    input_filename = "input.csv"
    output_filename = "output.csv"
    if len(sys.argv) > 1:
        input_filename = sys.argv[1]
        if len(sys.argv) > 2:
            output_filename = sys.argv[2]

    from CsvInnerJoiner import CsvInnerJoiner
    from SumJob import SumJob

    job = CsvInnerJoiner().run(input=[input_filename])
    job = SumJob().run() (******************)

    with open(output_filename, 'w') as fp:
        writer = csv.writer(fp)
        for url_key, descriptors in result_iterator(job.wait(show=True)):
            writer.writerow([url_key] + descriptors)

CsvInnerJoiner.py は次のファイルです。

import sys
sys.path.append("/home/damian/disco/lib/")
from disco.core import Job, result_iterator
from disco.worker.classic.func import chain_reader
import csv, sys
class CsvInnerJoiner(Job):
    partitions = 2
    sort = True

    def map(self, row, params):
        yield row[0], row[1:]

    @staticmethod
    def map_reader(fd, size, url, params):
        reader = csv.reader(fd, delimiter=',')
        for row in reader:
            yield row

    #@staticmethod
def reduce(self, rows_iter, out, params):
    from disco.util import kvgroup
    from itertools import chain
    #for url_key, descriptors in kvgroup(sorted(rows_iter)):
    for url_key, descriptors in kvgroup(rows_iter):
        merged_descriptors = list(chain.from_iterable(descriptors))
        print url_key,"_______",merged_descriptors
        if len(merged_descriptors) > 3:
            Alist = merged_descriptors[:merged_descriptors.index("B")]
            Blist = merged_descriptors[merged_descriptors.index("B"):]
            Alistlength = len(Alist)/3
            Blistlength = len(Blist)/3
            for i in range(Alistlength):
                for j in range(Blistlength):
                    container = int(Alist[3*i+2])*int(Blist[3*j+2])
                    yield [Alist[3*i+1],Blist[3*j+1]],container
                    #out.add(Alist[3*i+1],[Blist[3*j+1],container])

SumJob.py は次のとおりです。

import sys
sys.path.append("/home/damian/disco/lib/")
from disco.core import Job, result_iterator
from disco.worker.classic.func import chain_reader
import csv, sys


class SumJob(Job):
    map_reader = staticmethod(chain_reader)

    @staticmethod
    def map(self,key_value, params):
        print "KEY::::::",str(key_value[0])
        print "VAL::::::",str(key_value[1])
        yield key_value[0], key_value[1]
    @staticmethod    
    def reduce(self,key_value,out, params):
        Summe = sum(key_value[1])
        out.add(key_value[0],Summe)

問題は、最初の削減ステップの 2 番目の出力が 2 番目のマップ関数によって入力として取得されるように (**) 行を変更する方法がわからないことです。

手伝ってくれてどうもありがとう！ダミアン

4

1 に答える 1

0

map/reduce ステージの出力を別のステージの入力として使用できます (からの戻りjob.wait())。

job1 = SumJob().run(input=[...])
job2 = SumJob().run(input=[...])

output = SomeOtherJob.run(input=[job1.wait(), job2.wait()]).wait(show=True)
for key, value in result_iterator(output):
    print key, value

私はそのコードのチャンクが機能するという点で専門家ではありません（pagerank多くの段階といくつかの反復があるアルゴリズムを実装しています）。

于 2013-09-23T08:24:37.183 に答える