在大数据处理领域，Spark 和 HBase 以及 MySQL 都扮演着重要的角色。Spark 提供了高效的数据处理能力，HBase 是一个分布式、面向列的NoSQL数据库，而 MySQL 是广泛使用的的关系型数据库。本示例将详细介绍如何使用 Spark 从 HBase 中读取数据，并通过 Spark SQL 将其存储到 MySQL 数据库中。
让我们了解 Spark 与 HBase 的交互。Spark 提供了 `spark-hbase-connector` 库，允许我们方便地连接到 HBase 并进行数据操作。在 Scala 代码中，你需要先引入相应的依赖，例如在 `build.sbt` 或 `pom.xml` 文件中添加：
```scala
libraryDependencies += "com.hortonworks" %% "shc-core" % "1.1.1-2.1-s_2.11" // 对于 Scala 2.11 和 Spark 2.1
```
接着，创建一个 SparkSession，这是 Spark SQL 的入口点：
```scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
.appName("Spark-HBase-MySQL Integration")
.config("spark.master", "local[*]") // 根据实际情况设置
.getOrCreate()
```
然后，配置 HBase 连接参数，包括 `hbase.zookeeper.quorum`（Zookeeper 地址）和 `hbase.zookeeper.property.clientPort`（Zookeeper 端口）：
```scala
val conf = spark.sparkContext.hadoopConfiguration
conf.set("hbase.zookeeper.quorum", "zookeeper_host")
conf.set("hbase.zookeeper.property.clientPort", "2181") // 替换为实际值
```
现在，我们可以创建一个 DataFrame 来读取 HBase 中的数据：
```scala
import org.apache.spark.sql.execution.datasources.hbase.HBaseTableCatalog
val catalog = s"""{
"table":{"namespace":"default", "name":"my_table", "tableCoder":"PrimitiveType"},
"rowkey":"key",
"columns":{
"key":{"cf":"rowkey", "col":"key", "type":"string"},
"column1":{"cf":"data", "col":"col1", "type":"string"},
"column2":{"cf":"data", "col":"col2", "type":"string"}
}
}"""
val df = spark.read.options(Map(HBaseTableCatalog.tableCatalog -> catalog)).format("org.apache.spark.sql.execution.datasources.hbase").load()
```
读取完成后，你可以对数据进行处理，例如过滤、转换等。假设你已经处理并准备好了数据，接下来的目标是将这些数据存入 MySQL。确保已安装 `jdbc:mysql` 驱动，并配置 MySQL 连接：
```scala
import java.sql.DriverManager
val url = "jdbc:mysql://localhost:3306/mydb" // 替换为实际的数据库地址
val driver = "com.mysql.jdbc.Driver"
val username = "root" // 替换为实际的用户名
val password = "password" // 替换为实际的密码
Class.forName(driver)
```
接下来，创建一个 DataFrame 代表 MySQL 表结构，然后使用 `saveAsTable` 方法将数据写入：
```scala
import org.apache.spark.sql.jdbc.JdbcDialects$
val jdbcUrl = s"$url?useSSL=false&serverTimezone=UTC"
val properties = Map("user" -> username, "password" -> password)
df.write
.format("jdbc")
.option("url", jdbcUrl)
.option("dbtable", "my_table_in_mysql") // 替换为你的表名
.option("driver", driver)
.mode("append") // 可以根据需要更改为 "overwrite" 或 "errorifexists"
.properties(properties)
.save()
```
至此，我们就完成了从 HBase 读取数据并将其保存到 MySQL 的过程。这个过程的关键在于理解 Spark、HBase 和 MySQL 之间的交互机制，以及正确配置它们的连接参数。通过使用 DataFrame API 和 Spark SQL，可以方便地在不同的数据源之间进行数据迁移和处理。在实际应用中，根据具体需求，你可能还需要处理数据类型转换、错误处理等问题，以确保数据的一致性和完整性。
Demo.rar （1个子文件） 
Demo.scala 3KB
