我想你要找的是Spark流。spark有一个kafka连接器，可以链接到spark流上下文。
下面是一个非常基本的示例，它将在1分钟的时间间隔内为给定主题集中的所有消息创建rdd，然后按消息id字段对它们进行分组（您的值序列化程序必须公开这样一个 getMessageId 方法，当然）。

SparkConf conf = new SparkConf().setAppName(appName);
JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.minutes(1));

Map<String, Object> params = new HashMap<String, Object>() {{
    put("bootstrap.servers", kafkaServers);
    put("key.deserializer", kafkaKeyDeserializer);
    put("value.deserializer", kafkaValueDeserializer);
}};

List<String> topics = new ArrayList<String>() {{
    // Add Topics
}};

JavaInputDStream<ConsumerRecord<String, String>> stream =
    KafkaUtils.createDirectStream(ssc,
        LocationStrategies.PreferConsistent(),
        ConsumerStrategies.<String, String>Subscribe(topics, params)
    );

stream.foreachRDD(rdd -> rdd.groupBy(record -> record.value().getMessageId()));

ssc.start();
ssc.awaitTermination();

在流式api中还有其他几种方法可以对消息进行分组。有关更多示例，请参阅文档。