MRJob的多个输入

Question

我正在尝试学习将Yelp的Python
API用于MapReduce，MRJob。他们简单的单词计数器示例很有意义，但是我很好奇一个人如何处理涉及多个输入的应用程序。例如，将向量乘以矩阵，而不是简单地对文档中的单词进行计数。我想出了这个解决方案，它可以起作用，但是感觉很愚蠢：

class MatrixVectMultiplyTast(MRJob):
    def multiply(self,key,line):
            line = map(float,line.split(" "))
            v,col = line[-1],line[:-1]

            for i in xrange(len(col)):
                    yield i,col[i]*v

    def sum(self,i,occurrences):
            yield i,sum(occurrences)

    def steps(self):
            return [self.mr (self.multiply,self.sum),]

if __name__=="__main__":
    MatrixVectMultiplyTast.run()

该代码运行，./matrix.py < input.txt并且起作用的原因是，存储在input.txt中的矩阵按列存储，并且相应的向量值位于行的末尾。

因此，下面的矩阵和向量：

以input.txt表示为：

简而言之，我将如何更自然地将矩阵和向量存储在单独的文件中，然后将它们都传递给MRJob？

Answer 1

如果您需要针对另一个（或相同的row_i，row_j）数据集处理原始数据，则可以：

1）创建一个S3存储桶以存储数据副本。将此副本的位置传递到您的任务类，例如，以下代码中的self.options.bucket和self.options.my_datafile_copy_location。警告：不幸的是，似乎整个文件在处理之前必须先“下载”到任务计算机。如果连接变弱或加载时间太长，此作业可能会失败。这是一些Python
/ MRJob代码来执行此操作。

将其放在您的映射器函数中：

d1 = line1.split('\t', 1)
v1, col1 = d1[0], d1[1]
conn = boto.connect_s3(aws_access_key_id=<AWS_ACCESS_KEY_ID>, aws_secret_access_key=<AWS_SECRET_ACCESS_KEY>)
bucket = conn.get_bucket(self.options.bucket)  # bucket = conn.get_bucket(MY_UNIQUE_BUCKET_NAME_AS_STRING)
data_copy = bucket.get_key(self.options.my_datafile_copy_location).get_contents_as_string().rstrip()
### CAVEAT: Needs to get the whole file before processing the rest.
for line2 in data_copy.split('\n'):
    d2 = line2.split('\t', 1)
    v2, col2 = d2[0], d2[1]
    ## Now, insert code to do any operations between v1 and v2 (or c1 and c2) here:
    yield <your output key, value pairs>
conn.close()

2）创建一个SimpleDB域，并将所有数据存储在那里。在boto和SimpleDB上阅读此处：http
:
//code.google.com/p/boto/wiki/SimpleDbIntro

您的映射器代码如下所示：

dline = dline.strip()
d0 = dline.split('\t', 1)
v1, c1 = d0[0], d0[1]
sdb = boto.connect_sdb(aws_access_key_id=<AWS_ACCESS_KEY>, aws_secret_access_key=<AWS_SECRET_ACCESS_KEY>)
domain = sdb.get_domain(MY_DOMAIN_STRING_NAME)
for item in domain:
    v2, c2 = item.name, item['column']
    ## Now, insert code to do any operations between v1 and v2 (or c1 and c2) here:
    yield <your output key, value pairs>
sdb.close()

如果您有大量数据，则第二个选项可能会更好，因为它可以针对每一行数据而不是一次针对全部数据进行请求。请记住，SimpleDB值的最大长度不能超过1024个字符，因此，如果您的数据值比该值长，则可能需要通过某种方法进行压缩/解压缩。