Не удается получить доступ к данным в Кафки Спарк Потоковые глобально

Question

Я пытаюсь Потоковые данные из Кафки Спарк

JavaPairInputDStream<String, String> directKafkaStream = KafkaUtils.createDirectStream(ssc, 
       String.class, 
       String.class, 
       StringDecoder.class, 
       StringDecoder.class, 
       kafkaParams, topics); 

Здесь я итерация по JavaPairInputDStream для обрабатывать РДД-х.

directKafkaStream.foreachRDD(rdd ->{ 
      rdd.foreachPartition(items ->{ 
       while (items.hasNext()) { 
        String[] State = items.next()._2.split("\\,"); 
        System.out.println(State[2]+","+State[3]+","+State[4]+"--"); 
       }; 
      });   
     }); 

я могу иметь возможность получать данные в foreachRDD и мое требование иметь доступ к государству массива в глобальном масштабе. Когда я пытаюсь получить доступ к State Array во всем мире, я получаю исключение

java.lang.IndexOutOfBoundsException: Index: 0, Size: 0 

Любые предложения? Благодарю.

Answer 1

Это больше похоже на вашу таблицу поиска с потоковым RDD, чтобы получить все элементы, имеющие соответствующие поля «code» и «exceptionCode».

Поток должен быть таким.

1. Создание РДД из улья таблицы поиска => lookupRdd
2. Создание DStream от Кафки потока
3. Для каждого RDD в Dstream, присоединиться к lookupRDD с streamRdd, процесс соединяемых элементов (подсчитать сумму суммы ...) и сохраните этот обработанный результат.

Примечание Ниже код не заполнен. Пожалуйста, заполните все комментарии TODO.

JavaPairDStream<String, String> streamPair = directKafkaStream.mapToPair(new PairFunction<Tuple2<String, String>, String, String>() { 
     @Override 
     public Tuple2<String, String> call(Tuple2<String, String> tuple2) throws Exception { 
      System.out.println("Tuple2 Message is----------" + tuple2._2()); 
      String[] state = tuple2._2.split("\\,"); 
      return new Tuple2<>(state[4], tuple2._2()); //pair <ViolationCode, data> 
     } 
    }); 

    streamPair.foreachRDD(new Function<JavaPairRDD<String, String>, Void>() { 
     JavaPairRDD<String, String> hivePairRdd = null; 
     @Override 
     public Void call(JavaPairRDD<String, String> stringStringJavaPairRDD) throws Exception { 
      if (hivePairRdd == null) { 
       hivePairRdd = initHiveRdd(); 
      } 
      JavaPairRDD<String, Tuple2<String, String>> joinedRdd = stringStringJavaPairRDD.join(hivePairRdd); 
      System.out.println(joinedRdd.take(10)); 
      //todo process joinedRdd here and save the results. 
      joinedRdd.count(); //to trigger an action 
      return null; 
     } 
    }); 
} 

public static JavaPairRDD<String, String> initHiveRdd() { 
    JavaRDD<String> hiveTableRDD = null; //todo code to create RDD from hive table 
    JavaPairRDD<String, String> hivePairRdd = hiveTableRDD.mapToPair(new PairFunction<String, String, String>() { 
     @Override 
     public Tuple2<String, String> call(String row) throws Exception { 
      String code = null; //TODO process 'row' and get 'code' field 
      return new Tuple2<>(code, row); 
     } 
    }); 
    return hivePairRdd; 
}