java - Hadoop 矩阵乘法

Question

我正在运行在 http://www.norstad.org/matrix-multiply/index.html 找到的 MapReduce 矩阵乘法程序.我发现当输入矩阵中有 0 时，此实现无法正常工作。但我不明白为什么，我该如何修改程序才能使其工作？MapReduce 操作完成，但输出始终是一个所有元素为 0 的矩阵。

我的输入矩阵 A 和 B 是:

Matrix A     Matrix B
0 0 0        6 7 4 
0 1 6        9 1 3 
7 8 9        7 6 2  

输出矩阵:

0 0 0
0 0 0
0 0 0

以下内容取自作业的日志文件:

matrixB 的 map 输出:

##### Map setup: matrixA = false for hdfs://localhost/user/hadoop-user/B
strategy = 4
R1 = 4
I = 3
K = 3
J = 3
IB = 3
KB = 3
JB = 3
##### Map input: (0,0) 6
##### Map output: (0,0,0,1) (0,0,6) 
##### Map input: (0,1) 7
##### Map output: (0,0,0,1) (0,1,7) 
##### Map input: (0,2) 4
##### Map output: (0,0,0,1) (0,2,4) 
##### Map input: (1,0) 9
##### Map output: (0,0,0,1) (1,0,9) 
##### Map input: (1,1) 1
##### Map output: (0,0,0,1) (1,1,1) 
##### Map input: (1,2) 3
##### Map output: (0,0,0,1) (1,2,3) 
##### Map input: (2,0) 7
##### Map output: (0,0,0,1) (2,0,7) 
##### Map input: (2,1) 6
##### Map output: (0,0,0,1) (2,1,6) 
##### Map input: (2,2) 2
##### Map output: (0,0,0,1) (2,2,2) 

矩阵 A 的映射输出:

##### Map setup: matrixA = true for hdfs://localhost/user/hadoop-user/A
strategy = 4
R1 = 4
I = 3
K = 3
J = 3
IB = 3
KB = 3
JB = 3
##### Map input: (1,1) 1
##### Map output: (0,0,0,0) (1,1,1) 
##### Map input: (1,2) 6
##### Map output: (0,0,0,0) (1,2,6) 
##### Map input: (2,0) 7
##### Map output: (0,0,0,0) (2,0,7) 
##### Map input: (2,1) 8
##### Map output: (0,0,0,0) (2,1,8) 
##### Map input: (2,2) 9
##### Map output: (0,0,0,0) (2,2,9) 

请注意，矩阵 A 的映射不会从输入文件中读取零。注意:两个输入文件都作为序列文件存储在 HDFS 中，输出也是一个序列文件。有人可以阐明问题所在吗？谢谢!

Mapper类的代码如下:

/** The job 1 mapper class. */

private static class Job1Mapper 
    extends Mapper<IndexPair, IntWritable, Key, Value>
{
    private Path path;
    private boolean matrixA;
    private Key key = new Key();
    private Value value = new Value();

    public void setup (Context context) {
        init(context);
        FileSplit split = (FileSplit)context.getInputSplit();
        path = split.getPath();
        matrixA = path.toString().startsWith(inputPathA);
        if (DEBUG) {
            System.out.println("##### Map setup: matrixA = " + matrixA + " for " + path);
            System.out.println("   strategy = " + strategy);
            System.out.println("   R1 = " + R1);
            System.out.println("   I = " + I);
            System.out.println("   K = " + K);
            System.out.println("   J = " + J);
            System.out.println("   IB = " + IB);
            System.out.println("   KB = " + KB);
            System.out.println("   JB = " + JB);
        }
    }

    private void printMapInput (IndexPair indexPair, IntWritable el) {
        System.out.println("##### Map input: (" + indexPair.index1 + "," + 
            indexPair.index2 + ") " + el.get());
    }

    private void printMapOutput (Key key, Value value) {
        System.out.println("##### Map output: (" + key.index1 + "," + 
            key.index2 + "," + key.index3 + "," + key.m + ") (" + 
            value.index1 + "," + value.index2 + "," + value.v + ") ");
    }

    private void badIndex (int index, int dim, String msg) {
        System.err.println("Invalid " + msg + " in " + path + ": " + index + " " + dim);
        System.exit(1);
    }

    public void map (IndexPair indexPair, IntWritable el, Context context)
        throws IOException, InterruptedException 
    {
        if (DEBUG) printMapInput(indexPair, el);
        int i = 0;
        int k = 0;
        int j = 0;
        if (matrixA) {
            i = indexPair.index1;
            if (i < 0 || i >= I) badIndex(i, I, "A row index");
            k = indexPair.index2;
            if (k < 0 || k >= K) badIndex(k, K, "A column index");
        } else {
            k = indexPair.index1;
            if (k < 0 || k >= K) badIndex(k, K, "B row index");
            j = indexPair.index2;
            if (j < 0 || j >= J) badIndex(j, J, "B column index");
        }
        value.v = el.get();
                if (matrixA) {
                    key.index1 = i/IB;
                    key.index3 = k/KB;
                    key.m = 0;
                    value.index1 = i % IB;
                    value.index2 = k % KB;
                    for (int jb = 0; jb < NJB; jb++) {
                        key.index2 = jb;
                        context.write(key, value);
                        if (DEBUG) printMapOutput(key, value);
                    }
                } else {
                    key.index2 = j/JB;
                    key.index3 = k/KB;
                    key.m = 1;
                    value.index1 = k % KB;
                    value.index2 = j % JB;
                    for (int ib = 0; ib < NIB; ib++) {
                        key.index1 = ib;
                        context.write(key, value);
                        if (DEBUG) printMapOutput(key, value);
                    }
        }
    }
}

如果有任何语法错误，那只是因为我复制粘贴了代码并在此处进行了修改，所以我可能忽略了一些东西。

我需要有关 Map 函数逻辑的帮助，它不会从输入文件中读取 0，谁能告诉我为什么？

Answer 1

在 TestMatrixMultiply.java 中，从您链接的网站，它可能包含您用来将矩阵编码为预期的 IndexPair-backed 文件格式的代码，有函数 writeMatrix:

public static void writeMatrix (int[][] matrix, int rowDim, int colDim, String pathStr)
    throws IOException
{
    Path path = new Path(pathStr);
    SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, path, 
        MatrixMultiply.IndexPair.class, IntWritable.class, 
        SequenceFile.CompressionType.NONE);
    MatrixMultiply.IndexPair indexPair = new MatrixMultiply.IndexPair();
    IntWritable el = new IntWritable();
    for (int i = 0; i < rowDim; i++) {
        for (int j = 0; j < colDim; j++) {
            int v = matrix[i][j];
            if (v != 0) { // !!! well, that would be why we aren't writing 0s
                indexPair.index1 = i;
                indexPair.index2 = j;
                el.set(v);
                writer.append(indexPair, el);
            }
        }
    }
    writer.close();
}

注释插入到内部 for 循环的第二行。

您的映射器未读取任何 0，因为您的输入文件不包含任何 0。

该代码经过精心设计，假定所有缺失值均为 0，并且它会执行额外的检查以避免发出 0，以尝试最大程度地减少网络流量。

下面的大部分内容都是错误的，因为我误解了算法
(虽然上面的部分仍然有用)

在链接页面中，您正在使用策略 3。策略 3 依赖于分区程序行为和排序顺序。不幸的是，分区程序是错误的!这与未打印出来的 0 是分开的。分区程序在这里完全错误，你得到的矩阵全是 0，因为它乘以之前初始化为 0 的数据，并且从未被 block 的正确数据覆盖。这在 1 机操作中是隐藏的，因为分区程序是空操作，但在大多数集群中都会中断。

The partitioner maps intermediate key (kb, jb, ib) to a reducer r as follows:

r = (jb*KB + kb) mod R

需要保证同一个block的所有key都分配给同一个reducer。不幸的是，它保证这不会发生，除非 KB % numReducers == 0:

map (key, value)
   if from matrix A with key=(i,k) and value=a(i,k)
      for 0 <= jb < NJB
         emit (k/KB, jb, i/IB), (i mod IB, k mod KB, a(k,j)) // compare this...
   if from matrix B with key=(k,j) and value=b(k,j)
       emit (k/KB, j/JB, -1), (k mod KB, j mod KB, b(k,j))  // ...to this

对于矩阵 A，正在迭代键 jb。对于矩阵 B，正在计算 key jb。由于分配给 reducer 是循环的，这保证 A 矩阵键不会分配给与 B 矩阵键相同的 reducer。因此，该算法失败了，因为它假设了一些关于键的分配和排序的东西，这些东西可以证明是不正确的。当且仅当存在一个分配了所有键的 reducer 时，键顺序是正确的，但分区器是错误的!

分区器必须修改为使用策略 3 的 kb % numReducers。这不是一个很好的分区器，但它是唯一可以工作的分区器，因为需要不相同的键以特定顺序排序到同一个 reducer 。

您实际放入问题中的代码与错误实际存在的位置无关。