如何在Spark中按分区对键/值进行分组？

Question

我有一个Spark Streaming应用程序，每秒接收多个JSON消息，每个消息都有一个ID来标识它们的来源。

使用此ID作为键，我能够执行MapPartitionsToPair，从而创建一个JavaPairDStream，其中包含一个RDD的键值对，每个分区一个键值对（例如，如果我收到5个JSON消息，那么我将得到一个带有5个分区的RDD，每个分区的键是消息的ID，值是JSON消息本身）。

现在，我想将所有具有相同键的值分组到同一个分区中。因此，例如，如果我有3个键为'a'的分区和2个键为'b'的分区，我想创建一个新的RDD，其中有2个分区而不是5个分区，每个分区包含一个键的所有值，一个分区为'a'，另一个为'b'。

如何实现这一点？ 以下是我的代码：

JavaReceiverInputDStream<String> streamData2 = ssc.socketTextStream(args[0], Integer.parseInt(args[1]),
            StorageLevels.MEMORY_AND_DISK_SER);

JavaPairDStream<String,String> streamGiveKey= streamData2.mapPartitionsToPair(new PairFlatMapFunction<Iterator<String>, String, String>() {
        @Override
        public Iterable<Tuple2<String, String>> call(Iterator<String> stringIterator) throws Exception {

            ArrayList<Tuple2<String,String>>a= new ArrayList<Tuple2<String, String>>();

            while (stringIterator.hasNext()){
                String c=stringIterator.next();
                if(c==null){
                    return null;

                }

                JsonMessage retMap = new Gson().fromJson(c,JsonMessage.class);
                String key= retMap.getSid();
                Tuple2<String,String> b= new Tuple2<String,String>(key,c);
                a.add(b);

                System.out.print(b._1+"_"+b._2);
                // }
                //break;
            }


            return a;
        }
    });

//我创建了一个JavaPairDStream，其中每个分区都包含一个键/值对。

我尝试使用groupByKey()，但无论消息数量如何，我始终得到2个分区的数量。

我该怎么办？ 非常感谢。

- manuel mourato

---

你为什么想要每个分区只有1个元素？你试图解决什么问题？ - maasg

Answer 1

你可以使用HTML`

`元素来创建段落。

groupByKey(Integer numPartitions)

将numPartitions设置为您拥有的不同键的数量。

但是..你需要事先知道你有多少个不同的键。你有这个信息吗？可能没有。那么..你需要做一些额外（/冗余）的工作。例如，使用

countByKey

作为第一步，这比groupByKey更快 - 所以至少你没有使总处理时间加倍。

更新：OP问为什么默认情况下会得到2个分区。

默认的groupByKey使用defaultPartitioner()方法。

groupByKey(defaultPartitioner(self))

• 该功能从具有最大基数的父分区中选择Partitioner。

-- 或者将使用spark.default.parallelism

- WestCoastProjects

---

谢谢，这绝对解决了我的问题。不过，只有一个问题：你知道为什么 groupByKey() 默认返回2个分区吗？无论我每个批次间发送多少输入或输出，似乎groupByKey都与此无关。当我执行 getNumPartitions 时，它只返回2。 - manuel mourato