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

Question

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

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

所以如果我收到记录 (key, value_new), 我必须从 HBase (key, value_old)，所以我可以比较 value_new 和 value_old.

所以逻辑是：

Dstream from Kafka -> Query HBase by DStream keys -> Some calculations -> Write to HBase

我的“幼稚”方法是使用 Phoenix Spark 连接器读取并根据键左连接到新数据，以此过滤掉不在当前微批中的键。所以我会得到一个带有 (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);

    }

Answer 1

解决方案是使用spark hbase connector进行批量get和put。

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

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

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

Spark Streaming: Read from HBase by received stream keys?

hbase

apache-kafka

apache-spark

spark-streaming

spark-streaming-kafka