創建Mysql數據庫

CREATE DATABASE spark; USE spark;CREATE TABLE DEPT( DEPTNO int(2) PRIMARY KEY, DNAME VARCHAR(14), LOC VARCHAR(13) );INSERT INTO DEPT VALUE(10, 'ACCOUNTING','NEW YORK'); INSERT INTO DEPT VALUE(20, 'RESEARCH','DALAS'); INSERT INTO DEPT VALUE(30, 'SALES','CHICAGO'); INSERT INTO DEPT VALUE(40, 'OPERATIONS','BOSTON');

創建Hive數據庫

CREATE TABLE emp( empno int, ename string, job string, mgr int, hiredate string, sal double, comm double, deptno int ) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

準備文本數據emp.txt：

1 tom clerk 9088 1980-12-09 800.0 NULL 20 2 vincent cl 9999 1992-03-04 1000.0 300.1 30 5 sofia salesman 8000 1996-02-22 100.0 908 20

將數據導入到hive中：
hive> load data local inpath '/home/iie4bu/data/emp.txt' overwrite into table emp;

JOIN操作

package cn.ac.iie.sparkimport org.apache.spark.sql.SparkSession/*** 使用外部數據源綜合查詢Hive和Mysql的表數據*/ object HiveMysqlApp {def main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("HiveMysqlApp").master("local[2]").getOrCreate()// 加載hive表數據val hiveDF = spark.table("emp")// 加載mysql數據val mysqlDF = spark.read.format("jdbc").option("url", "jdbc:mysql://swarm-manager:3306").option("driver", "com.mysql.jdbc.Driver").option("dbtable", "spark.DEPT").option("user", "root").option("password", "123456").load()// JOINval resultDF = hiveDF.join(mysqlDF,hiveDF.col("deptno") === mysqlDF.col("DEPTNO")) resultDF.select(hiveDF.col("empno"),hiveDF.col("ename"),mysqlDF.col("deptno"),mysqlDF.col("dname")).show()resultDF.show()spark.close()} }

代碼需要導入mysql驅動jars ，所以在spark-shell中執行：

總結

以上是生活随笔為你收集整理的SparkSQL之关联mysql和hive查询的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。