Zuordnung von Schlüssel/Wert-Partitionen parallel in Spark Streaming

Question

Ich habe ein Spark-Streaming-Programm im lokalen Modus, in dem ich JSON-Nachrichten von einer TCP-Socket-Verbindung erhalten, mehrere pro Batch-Intervall.

Jede dieser Nachrichten hat eine ID, mit der ich einen Schlüssel/Wert JavaPairDStream erzeuge, so dass es in jeder Partition der RDD in meinem JavaDStream ein Schlüssel/Wert-Paar mit einer einzigen Nachricht pro Partition gibt.

Mein Ziel ist es nun, die Nachrichten, die dieselbe ID haben, in derselben Partition zu gruppieren, so dass ich sie parallel abbilden kann, wobei jede Partition von einem anderen Kern bearbeitet wird.

Im Anschluss ist mein Code:

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

JavaPairDStream<String,String> streamGiveKey= streamData1.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); 

      } 

      return a; 
     } 
    }); 

am Ende dieses Codes Also, ich habe eine DSTREAM mit einem RDD, die wegen repartition{1} nur eine Partition hat, mit den alle Schlüssel/Wert-Paare im Inneren .

Wie soll ich jetzt fortfahren, um die Nachrichten, die den gleichen Schlüssel haben, zu gruppieren und sie in verschiedene Partitionen zu legen, damit ich sie getrennt abbilden kann?

Answer 1

Diese Frage wurde - nach der OP - angesprochen durch eine gleichzeitige Antwort auf eine andere Frage hier: How to group key/values by partition in Spark?