在 Hadoop 的 Reducer 中获取整数数组

Question

我正在调用一个 cuda 代码来获取每个键的所有值的总和。 目的是通过并行操作来减少reducer所花费的时间。 但是，reducer 中的值是 IntWritable 形式。因此，我必须将它们转换为整数数组以传递给 cuda 代码。 这是我的减速器代码：

public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
   private IntWritable result = new IntWritable();

   public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
       List<Integer> numbers = new ArrayList<Integer>();
       for(IntWritable val : values)
            numbers.add(val.get());
       }
       int[] ret = ArrayUtils.toPrimitive(numbers.toArray(new Integer[numbers.size()]));
       result.set(Main.sumNumbers(ret));
       context.write(key,result);
   }
}

问题在于，为了将 IntWritable 转换为 Integer 数组，我必须遍历每个作为串行操作的值。因此，它正在增加更多的时间。 那么，有什么方法可以让我不必遍历每个值并直接转换为 int 数组？

这是映射器代码：

public static class TokenizerMapper extends
            Mapper<Object, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }

这是我的 cuda 代码：

#include <stdio.h>
#ifndef _ADDARRAY_KERNEL_H_
#define _ADDARRAY_KERNEL_H_

#ifdef __cplusplus
extern "C"
{
#endif
__global__ void add_array(int *a, int *c, int N)
{
  *c = 0;
  int i;
   for(i = 0; i<N;i++)
   {
    *c = *c + a[i];
   }
}
#ifdef __cplusplus
}
#endif 
#endif // #ifndef _ADDARRAY_KERNEL_H_

#ifdef __cplusplus
extern "C"
{
#endif

int cuda_sum(int *a_h, int N)
{   
    int *a_d, c=0;
    int *dev_c;
    cudaMalloc((void**)&dev_c, sizeof(int));
    size_t size = N * sizeof (int);

//      a_h = (int *) malloc(size);
    cudaMalloc((void **) & a_d, size);
    cudaMemcpy(a_d, a_h, size, cudaMemcpyHostToDevice);
        add_array <<<1, 1 >>>(a_d, dev_c, N);
        cudaMemcpy(&c, dev_c, sizeof(int), cudaMemcpyDeviceToHost);
    cudaFree(dev_c);
    return c;
}
#ifdef __cplusplus
}
#endif 

谢谢

Answer 1

我建议你应该做如下的事情

public static class IntSumReducer extends Reducer<Text, IntWritable, Text, ArrayPrimitiveWritable>{ private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<ArrayPrimitiveWritable> values, Context context) throws IOException,InterruptedException { int[] ret = values.next(); result.set(Main.sumNumbers(ret)); context.write(key,result); } }

ArrayPrimitiveWritable 将为您完成这项工作。

Answer 2

public static class IntSumReducer extends Reducer<Text, ArrayPrimitiveWritable, Text, IntWritable>{
        private IntWritable result = new IntWritable();
        public void reduce(Text key, Iterable<ArrayPrimitiveWritable> values, Context context) throws IOException,InterruptedException {
            ArrayPrimitiveWritable arrayOfInts = values.iterator().next();
            final int[] ret = (int [])arrayOfInts.get();
            result.set(Main.sumNumbers(ret));
            context.write(key,result);
        }
    }

我已经稍微修改了代码以便它可以工作，但是我还没有运行它，你可以试一试吗？我坚持使用刚刚更新的 ArrayPrimitiveWritable，这样编译错误就会消失。