創(chuàng)建Mysql數(shù)據(jù)庫(kù)

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');

創(chuàng)建Hive數(shù)據(jù)庫(kù)

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';

準(zhǔn)備文本數(shù)據(jù)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

將數(shù)據(jù)導(dǎo)入到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/*** 使用外部數(shù)據(jù)源綜合查詢Hive和Mysql的表數(shù)據(jù)*/ object HiveMysqlApp {def main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("HiveMysqlApp").master("local[2]").getOrCreate()// 加載hive表數(shù)據(jù)val hiveDF = spark.table("emp")// 加載mysql數(shù)據(jù)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()} }

代碼需要導(dǎo)入mysql驅(qū)動(dòng)jars ，所以在spark-shell中執(zhí)行：

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。