Java spark 使用 reduceByKey 避免嵌套列表将对象连接到一个列表中

Question

我有一个 java spark rdd，其中（键，值）对作为（String，ArrayList）。 ArrayList 是一个 Object 数组（可以是 int、double、string 或 array 或任何其他类型）。

例如输入如下：

key1, [R1_Entry_1, R1_Entry_2, ..., R1_Entry_n]
key1, [R2_Entry_1, R2_Entry_2, ..., R2_Entry_n]
key1, [R3_Entry_1, R3_Entry_2, ..., R3_Entry_n]
key2, [R4_Entry_1, R4_Entry_2, ..., R4_Entry_n]
... 
keyJ, [RK_Entry_1, RK_Entry_2, ..., RK_Entry_n]
... 
keyX, [RM_Entry_1, RM_Entry_2, ..., RM_Entry_n]

那么 RI_entry_J（第 I 行和第 J 行）是一个 java.lang.Object，可以是 int、double、字符串、ArrayList 或任何其他类型。

我使用了 JavaRDD.reduceByKey() 来减少进入新列表的条目。我想要的输出应该是（假设 key1 只有三个对应的行（前三行：R1、R2、R3）。）

key1, [[R1_Entry_1, R2_Entry_1,R3_Entry_1], [R1_Entry_2, R2_Entry_2,R3_Entry_2], ..., [R1_Entry_n, R2_Entry_n,R3_Entry_n]]
... 
keyJ, [[RK_Entry_1, R*_Entry_1, ...], [RK_Entry_2, R*_Entry_2, ...], ..., [RK_Entry_n, R*_Entry_n, ...]]
...

我的代码如下：

JavaPairRDD<String, ArrayList> AdjJavaRDD =  JavaRDD.reduceByKey(new Function2<ArrayList, ArrayList, ArrayList>() {
@Override
public ArrayList call(ArrayList v1, ArrayList v2) throws Exception {

    int v1Len = v1.size();
    int v2Len = v2.size();
    if (v1Len != v2Len) {
        System.out.println(" \n The input size is incorrect. Please check! \n  ");
        System.exit(0);
    }

    List<Object> obj = new ArrayList<Object>(v1Len);
    for (int i =0; i < v1Len; i++)
    {
        List<Object> obj_i = new ArrayList<>();
        Object v1i = v1.get(i);
        Object v2i = v2.get(i);
        obj_i.add(v1i);
        obj_i.add(v2i);
        obj.add(i, obj_i);
    }
    return new ArrayList(obj);
}
});

我得到的结果如下： key1, [[[R1_Entry_1, R2_Entry_1],R3_Entry_1], [[R1_Entry_2, R2_Entry_2],R3_Entry_2], ..., [[R1_Entry_n, R2_Entry_n],R3_Entry_n]]

即条目嵌套在列表中，而不是写入没有嵌套的列表。

例如，如果 Entry_1 是整数，并且 KeyJ 有五个对应的行，其中五个 Entry_1 为 1, 2, 3,4, 5。 KeyJ 使用我的代码的结果是 KeyJ, [[[[[1,2],3], 4], 5], ...]（实际计算时，整数的顺序是随机的）。然而，我想要的是 KeyJ, [[1, 2, 3, 4, 5], ...]

有什么想法可以在 Java 的 Spark 代码中避免这种嵌套列表吗？

Answer 1

我有办法。上面的代码不起作用的原因是 Object v1i = v1.get(i); 将 return 一个特定的对象，例如v1i 首次调用时作为 String （或任何其他数据类型）。但是，当它之后被调用时，v1i 变成了 ArrayList，这就是问题 return 嵌套列表中的代码的原因。

更新后的解决方案如下：

  JavaPairRDD<Tuple2, ArrayList> JavaRDDColumns = adjJavaRDD.reduceByKey(new Function2<ArrayList, ArrayList, ArrayList>() {
            @Override
            public ArrayList call(ArrayList v1, ArrayList v2) throws Exception {
                int v1Len = v1.size();
                int v2Len = v2.size();

                Object[] objArr = new Object[v1Len];
                for (int i = 0; i < v1Len; i++) {
                    ArrayList<Object> obj_i = new ArrayList<Object>();
                    Object obj1i = v1.get(i);
                    Object obj2i = v2.get(i);
                    List<Object> obj1Arr = new ArrayList<>();
                    List<Object> obj2Arr = new ArrayList<>();
                    if (obj1i instanceof List<?>) {
                        obj1Arr = (List) obj1i;
                    } else {
                        obj1Arr = Arrays.asList(obj1i);  // not a Arraylist, convert to a Arraylist
                    }

                    if (obj2i instanceof List<?>) {
                        obj2Arr = (List) obj2i;
                    } else {
                        obj2Arr = Arrays.asList(obj2i);
                    }
                    List<Object> combineList = ListUtils.union(obj1Arr, obj2Arr);
                    objArr[i] = combineList;
                }
                ArrayList<Object> obj = new ArrayList<>(Arrays.asList(objArr));
                return obj;
            }
        });

Java spark 使用 reduceByKey 避免嵌套列表将对象连接到一个列表中

Java spark concatenate object into one list using reduceByKey aviod nested list

java

arrays

reduce

object

apache-spark