Spark Streaming：通过接收到的流密钥从 HBase 读取？

如何解决Spark Streaming：通过接收到的流密钥从 HBase 读取？

将 Spark Streaming 中接收到的数据与 HBase 中的现有数据进行比较的最佳方法是什么？

我们从 kafka 接收数据作为 DStream，在将其写入 HBase 之前，我们必须根据从 kafka 接收到的键扫描 HBase 中的数据，做一些计算（基于每个键的新数据和旧数据） ) 然后写到 HBase。

因此，如果我收到记录 (key,value_new)，我必须从 HBase (key,value_old) 获取，以便我可以比较 value_new 与 value_old。

所以逻辑是：

来自Kafka的Dstream -> 通过DStream键查询HBase -> 一些计算 -> 写入 HBase

我的“天真”方法是使用 Phoenix Spark Connector 读取并根据键对新数据进行左连接，以此过滤掉不在当前微批次中的键。所以我会得到一个带有 (key,value_new,value_old) 的 DF，从这里我可以比较内部分区。

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 ...

});

问题：使用上述方法根据收到的 DStream RDD 中的键列表从 HBase 获取数据对于流式传输来说太慢了。

有什么高效的方法可以做到这一点？

---

旁注： 方法 toPhoenixTableDF 只是将接收到的 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);

    }

解决方法

解决方案是使用 spark hbase 连接器进行批量获取和放置。

您可以在此处找到带有很好示例的源代码。 https://github.com/apache/hbase-connectors/tree/master/spark 以及 HBase 文档（火花会话）。

这个库使用普通的 Java/Scala Hbase api，所以你可以控制操作，但是通过广播给执行程序的 hbaseContext 对象为你管理连接池，这真的很棒。它为 Hbase 操作提供了简单的包装器，但是，如果需要，我们可以只使用它的 foreach/mapPartition 并获得对逻辑的控制，同时可以访问托管连接。