集成MongoDB：HBase与MongoDB的集成和应用

import org.apache.spark.sql.SparkSession
import com.mongodb.spark.config.ReadConfig
import com.mongodb.spark.rdd.MongoRDD
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.hbase.HBaseTableCatalog
import org.apache.spark.sql.hbase.HBaseUtils._
 
// 初始化Spark会话
val spark = SparkSession.builder()
  .appName("MongoDBHBaseIntegration")
  .config("spark.mongodb.input.uri", "mongodb://localhost:27017/database.collection")
  .getOrCreate()
 
// 从MongoDB读取数据
val readConfig = ReadConfig(Map("collection" -> "collection"))
val mongoRDD = MongoRDD.builder()
  .sparkContext(spark.sparkContext)
  .readConfig(readConfig)
  .build()
 
// 将RDD转换为DataFrame
val df = spark.read.json(mongoRDD)
 
// 将DataFrame注册为临时表
df.createOrReplaceTempView("mongo_table")
 
// 将数据写入HBase
val catalog = s"""{
  |  "table":{
  |    "namespace":"hbase_namespace",
  |    "name":"hbase_table",
  |    "tableCoder":"PrimitiveType"
  |  },
  |  "columns":{
  |    "id":{
  |      "qualifier":"id",
  |      "type":"binary"
  |    },
  |    "name":{
  |      "qualifier":"name",
  |      "type":"string"
  |    },
  |    // ... 其他列映射
  |  }
  |}""".stripMargin
 
// 将DataFrame保存到HBase
df.write.mode(SaveMode.Append).options(Map("catalog" -> catalog)).format("org.apache.spark.sql.execution.datasources.hbase").save()

这段代码展示了如何使用Spark SQL从MongoDB读取数据，并将其保存到HBase中。首先，它配置了Spark会话并从MongoDB读取数据。接着，它将数据转换为DataFrame，并创建一个HBase的catalog配置，最后使用DataFrame的save方法将数据保存到HBase中。这个过程是整合MongoDB和HBase的一个很好的例子。