如何在 Spark Kafka 连接器中获取对象的 JavaDStream？

Question

我正在使用 Spark Kafka 连接器从 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;
}

我的目标是将这些数据保存到 Cassandra 中，其中的表与 EventLog 具有相同的规格。 Spark Cassandra 连接器在插入语句中接受JavaRDD<EventLog>，如下所示：javaFunctions(rdd).writerBuilder("ks", "event", mapToRow(EventLog.class)).saveToCassandra();。我想从 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();
  }
});

EventLogDecoder 应该实现 kafka.serializer.Decoder。下面是 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;
 }
}