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

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

比较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),从这里我可以比较内部分区。

  1. JavaInputDStream<ConsumerRecord<String, String>> kafkaDStream = KafkaUtils.createDirectStream(...);
  2. // use foreachRDD in order to use Phoenix DF API
  3. kafkaDStream.foreachRDD((rdd, time) -> {
  4. // Get the singleton instance of SparkSession
  5. SparkSession spark = JavaSparkSessionSingleton.getInstance(rdd.context().getConf());
  6. JavaPairRDD<String, String> keyValueRdd = rdd.mapToPair(record -> new Tuple2<>(record.key(), record.value()));
  7. // TO SLOW FROM HERE
  8. Dataset<Row> oldDataDF = spark
  9. .read()
  10. .format("org.apache.phoenix.spark")
  11. .option("table", PHOENIX_TABLE)
  12. .option("zkUrl", PHOENIX_ZK)
  13. .load()
  14. .withColumnRenamed("JSON", "JSON_OLD")
  15. .withColumnRenamed("KEY_ROW", "KEY_OLD");
  16. Dataset<Row> newDF = toPhoenixTableDF(spark, keyValueRdd); //just a helper method to get RDD to DF (see note bellow)
  17. Dataset<Row> newAndOld = newDF.join(oldDataDF, oldDataDF.col("KEY_OLD").equalTo(newDF.col("KEY_ROW")), "left");
  18. /// do some calcs based on new vs old values and then write to Hbase ...
  19. });

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

  1. private static Dataset<Row> toPhoenixTableDF(SparkSession spark, JavaPairRDD<String, String> keyValueRdd) {
  2. JavaRDD<phoenixTableRecord> tmp = keyValueRdd.map(x -> {
  3. phoenixTableRecord record = new phoenixTableRecord();
  4. record.setKEY_ROW(x._1);
  5. record.setJSON(x._2);
  6. return record;
  7. });
  8. return spark.createDataFrame(tmp, phoenixTableRecord.class);
  9. }
p5cysglq

p5cysglq1#

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

相关问题