Как получить данные об использовании, что поступающие от Кафки, используя искру

Question

я написал ниже код, чтобы потреблять данные с помощью искровой Работа,

Есть ли что-нибудь не хватает для потокового Кафки или обработки данных, после того, как получил получить? Как я могу проверить, что данные извлекаются или нет?

// StreamingExamples.setStreamingLogLevels(); 
     SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount").setMaster("local[*]"); 
     ; 
     // Create the context with 2 seconds batch size 
     JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(1000)); 

     Map<String, Integer> topicMap = new HashMap<>(); 
     topicMap.put("Ptopic", 1); 

     JavaPairReceiverInputDStream<String, String> messages = KafkaUtils.createStream(jssc, "localhost:2181", "5", 
       topicMap); 

     /*messages.foreach(new Function<JavaRDD<String, String>, Void>() { 
      public Void call(JavaRDD<String, String> accessLogs) { 
        return null; 
       }} 
      );*/ 

     JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() { 
      @Override 
      public String call(Tuple2<String, String> tuple2) { 
       /*System.out.println(tuple2._1().toString()); 
       System.out.println(tuple2._2().toString());*/ 
       return tuple2._2(); 
      } 
     }); 

     lines.print(); 
     jssc.start(); 
     jssc.awaitTermination(); 

Вот результат только печать ..

Answer 1

Вы можете использовать основные функции больших данных, как карты, уменьшить, flatmap и так далее.

Update 1:

JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() { 
     @Override 
     public String call(Tuple2<String, String> tuple2) { 
      /*System.out.println(tuple2._1().toString()); 
      System.out.println(tuple2._2().toString());*/ 
      return tuple2._2(); 
     } 
    }); 

    // TODO: make some transformation here: 
    lines = lines.map(x -> { // clean data 
     String callType = x.getCallType().replaceAll("\"", "").replaceAll("[-|,]", ""); // here some operations 
     x.setCallType(callType); 
     return x; 
    }).filter(pair -> { // filter data 
     return !isFilteredOnFire || pair.getCallType().matches("(?i).*\\bFire\\b.*"); // here so filters 
    }); 

    lines.print(); 
    jssc.start(); 
    jssc.awaitTermination(); 

Полный пример описан в this blog post