Spark Kafka ConnectorでオブジェクトのJavaDStreamを取得するには？

Question

私はKafkaクラスタからデータを取得するためにSpark Kafkaコネクタを使用しています。それから私はJavaDStream<String>としてデータを取得しています。データをJavaDStream<EventLog>として取得するには、EventLogはJava Beanですか？

public static JavaDStream<EventLog> fetchAndValidateData(String zkQuorum, String group, Map<String, Integer> topicMap) { 
    SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount"); 
    JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000)); 
    JavaPairReceiverInputDStream<String, String> messages = 
      KafkaUtils.createStream(jssc, zkQuorum, group, topicMap); 
    JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() { 
     @Override 
     public String call(Tuple2<String, String> tuple2) { 
      return tuple2._2(); 
     } 
    }); 
    jssc.start(); 
    jssc.awaitTermination(); 
    return lines; 
} 

私の目標は、EventLogと同じ仕様のテーブルカサンドラにこのデータを保存することです。 Spark Cassandraコネクタは、javaFunctions(rdd).writerBuilder("ks", "event", mapToRow(EventLog.class)).saveToCassandra();のようにinsert文にJavaRDD<EventLog>を受け入れます。私はKafkaからJavaRDD<EventLog>を手に入れたいです。

Answer 1

は、キー/値の型とデコーダクラスを渡すことができ、オーバーロードcreateStreamメソッドを使用します。

例：

は

createStream(jssc, String.class, EventLog.class, StringDecoder.class, EventLogDecoder.class, 
     kafkaParams, topicsMap, StorageLevel.MEMORY_AND_DISK_SER_2()); 

以上があなたJavaPairDStream<String, EventLog>

JavaDStream<EventLog> lines = messages.map(new Function<Tuple2<String, EventLog>, EventLog>() { 
    @Override 
    public EventLog call(Tuple2<String, EventLog> tuple2) { 
    return tuple2._2(); 
    } 
}); 

にkafka.serializer.Decoderを実装する必要がありEventLogDecoderを与える必要があります。以下はjsonデコーダの例です。

public class EventLogDecoder implements Decoder<EventLog> { 

public EventLogDecoder(VerifiableProperties verifiableProperties) { 
} 

@Override 
public EventLog fromBytes(byte[] bytes) { 
    ObjectMapper objectMapper = new ObjectMapper(); 
    try { 
    return objectMapper.readValue(bytes, EventLog.class); 
    } catch (IOException e) { 
    //do something 
    } 
    return null; 
} 
}