比较spark流中接收的数据与hbase中现有数据的最佳方法是什么？
我们以数据流的形式从kafka接收数据，在将数据写入hbase之前，我们必须根据从kafka接收到的密钥扫描hbase，进行一些计算（基于每个密钥的新旧数据），然后写入hbase。
因此，如果我收到记录（key，value\u new），我必须从hbase（key，value\u old）获取，这样我就可以比较value\u new和value\u old。
所以逻辑是：
Kafka的数据流->通过数据流键查询hbase->一些计算->写入hbase
我的“娜”ï“ve”的方法是使用phoenix spark连接器读取并左键连接到基于键的新数据，以此过滤掉不在当前微批中的键。所以我会得到一个df（key，value\u new，value\u old），从这里我可以比较内部分区。

JavaInputDStream<ConsumerRecord<String, String>> kafkaDStream = KafkaUtils.createDirectStream(...);

// use foreachRDD in order to use Phoenix DF API
kafkaDStream.foreachRDD((rdd, time) -> {
        // Get the singleton instance of SparkSession
        SparkSession spark = JavaSparkSessionSingleton.getInstance(rdd.context().getConf());

        JavaPairRDD<String, String> keyValueRdd = rdd.mapToPair(record -> new Tuple2<>(record.key(), record.value()));

        // TO SLOW FROM HERE
        Dataset<Row> oldDataDF = spark
                .read()
                .format("org.apache.phoenix.spark")
                .option("table", PHOENIX_TABLE)
                .option("zkUrl", PHOENIX_ZK)
                .load()
                .withColumnRenamed("JSON", "JSON_OLD")
                .withColumnRenamed("KEY_ROW", "KEY_OLD");

        Dataset<Row> newDF = toPhoenixTableDF(spark, keyValueRdd); //just a helper method to get RDD to DF (see note bellow)

        Dataset<Row> newAndOld = newDF.join(oldDataDF, oldDataDF.col("KEY_OLD").equalTo(newDF.col("KEY_ROW")), "left");

        /// do some calcs based on new vs old values and then write to Hbase ...

});

问题：使用上述方法从接收到的数据流rdd中基于密钥列表从hbase获取数据对于流传输来说太慢。
有什么好办法可以做到这一点？
旁注：方法tophoenix tabledf只是将接收到的rdd转换为df的助手：

private static Dataset<Row> toPhoenixTableDF(SparkSession spark, JavaPairRDD<String, String> keyValueRdd) {
        JavaRDD<phoenixTableRecord> tmp = keyValueRdd.map(x -> {
            phoenixTableRecord record = new phoenixTableRecord();
            record.setKEY_ROW(x._1);
            record.setJSON(x._2);
            return record;
        });

        return spark.createDataFrame(tmp, phoenixTableRecord.class);

    }