spark流:通过接收到的流密钥从hbase读取?

ldxq2e6h  于 2021-07-13  发布在  Spark
关注(0)|答案(1)|浏览(378)

比较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);

    }
p5cysglq

p5cysglq1#

解决方案是使用spark hbase连接器进行批量获取和放置。
你可以在这里找到源代码和很好的例子。https://github.com/apache/hbase-connectors/tree/master/spark 以及hbase文档(spark会话)。
这个库使用普通的java/scala hbase api,因此您可以控制操作,但是可以通过广播给执行者的hbasecontext对象来管理连接池,这非常好。它为hbase操作提供了简单的 Package 器,但是,如果需要,我们可以只使用它的foreach/mappartition并获得对逻辑的控制,同时可以访问托管连接。

相关问题