Spark SQL配置及使用教程

目录

• SparkSQL版本： 

• SparkSQL DSL语法 

• Spark中beeline的使用
• 通过jdbc来访问spark的ThriftServer接口

• 案例一：SparkSQL读取HDFS上Json格式的文件
• 案例二：DataFrame和Dataset和RDD之间的互相转换

$ ln -s /opt/modules/apache/hive-1.2.1/conf/hive-site.xml 
or
$ cp /opt/modules/apache/hive-1.2.1/conf/hive-site.xml ./

spark-sql (default)> select * from emp;

spark-sql (default)> select a.*,b.* from emp a left join dept b on a.deptno = b.deptno;

> cache table emp;    #缓存操作
> uncache table dept;    #清除缓存操作
> explain select * from emp;    #执行计划

$ bin/spark-shell
#可以在shell里面写sql
scala> spark.sql("show databases").show
scala> spark.sql("use common").show
scala> spark.sql("select * from emp a join dept b on a.deptno = b.deptno").show

scala> df.registerTempTable("table_regis01")

<!-- 监听的端口号 -->
<property>
	<name>hive.server2.thrift.bind.port</name>
	<value>10000</value>
</property>
<!-- 监听的主机名 -->
<property>
	<name>hive.server2.thrift.bind.host</name>
	<value>hadoop01.com</value>
</property>

$ ./bin/hive --service metastore &

$ sbin/start-thriftserver.sh 

$ bin/beeline    #启动beeline
#可以使用!help查看相应的命令
beeline> !help
#如connect
beeline> !connect
Usage: connect <url> <username> <password> [driver]
#这样可以多个用户连接
beeline> !connect jdbc:hive2://hadoop01.com:10000
#退出
beeline> !quit

package com.jeffrey
import java.sql.DriverManager
object SparkJDBCThriftServerDemo {
    def main(args: Array[String]): Unit = {
        //1 添加驱动
        val driver = "org.apache.hive.jdbc.HiveDriver"
        Class.forName(driver)
        //2 构建连接对象
        val url = "jdbc:hive2://hadoop01.com:10000"
        val conn = DriverManager.getConnection(url,"ijeffrey","123456")
        //3 sql 语句执行
        conn.prepareStatement("use common").execute()
        var pstmt = conn.prepareStatement("select empno,ename,sal from emp")
        var rs = pstmt.executeQuery()
        while (rs.next()){
            println(s"empno = ${rs.getInt("empno")}  " +
                    s"ename=${rs.getString("ename")}   " +
                    s" sal=${rs.getDouble("sal")}")
        }
        println("---------------------------------------------------------------------------")
        pstmt = conn.prepareStatement("select empno,ename,sal from emp where sal > ? and ename = ?")
        pstmt.setDouble(1,3000)
        pstmt.setString(2,"KING")
        rs = pstmt.executeQuery()
        while (rs.next()){
            println(s"empno = ${rs.getInt("empno")}  " +
                    s"ename=${rs.getString("ename")}   " +
                    s" sal=${rs.getDouble("sal")}")
        }
        rs.close()
        pstmt.close()
        conn.close()
    }
}

scala> val path = "/spark/data/people.json"
scala> val df = spark.read.json(path)
scala> df.registerTempTable("tmp04") //通过DataFrame注册一个临时表
scala> spark.sql("show tables").show  //通过SQL语句进行操作
scala> spark.sql("select * from tmp04").show
#saveAsTable 使用之前 先要use table
scala> spark.sql("select * from tmp04").write.saveAsTable("test01")
#overwrite 覆盖  append 拼接  ignore 忽略
scala> spark.sql("select * from tmp01").write.mode("overwrite").saveAsTable("test01")
scala> spark.sql("select * from tmp01").write.mode("append").saveAsTable("test01")
scala> spark.sql("select * from tmp01").write.mode("ignore").saveAsTable("test01")

scala> spark.sql("select * from json.`/spark/data/people.json` where age is not null").show 
+---+------+
|age|  name|
+---+------+
| 30|  Andy|
| 19|Justin|
+---+------+
# hdfs上的路径使用`(反票号)引起来

package com.jeffrey.sql
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
object HiveJoinMySQLDemo {
    def main(args: Array[String]): Unit = {
        System.setProperty("hadoop.home.dir","D:\\hadoop-2.7.3")
        // 1.构建SparkSession
        val warehouseLocation = "/user/hive/warehouse"
        val spark = SparkSession
                .builder()
                .master("local")    //如果放到集群运行需要注释掉
                .appName("RDD 2 DataFrame")
                .config("spark.sql.warehouse.dir",warehouseLocation)
                .enableHiveSupport()
                .getOrCreate()
        import spark.implicits._
        import spark.sql
        val url = "jdbc:mysql://hadoop01.com:3306/test"
        val table = "tb_dept"
        val props = new Properties()
        props.put("user","root")
        props.put("password","123456")
        // 1.Hive表数据导入到MySQL中    在shell中可以使用paste写多行
        spark.read.table("common.dept")
                .write
                .mode(SaveMode.Overwrite)
                .jdbc(url,table,props)
        // 2.Hive和MySQL的join操作
        //2.1 读取MySQL的数据
       val df: DataFrame = spark
                .read
                .jdbc(url,table,props)
        df.createOrReplaceTempView("tmp_tb_dept")
        //2.1 数据聚合
        spark.sql(
            """
              |select a.*,b.dname,b.loc
              |from common.emp a
              |join tmp_tb_dept b on a.deptno = b.deptno
            """.stripMargin).createOrReplaceTempView("tmp_emp_join_dept_result")
        spark.sql("select * from tmp_emp_join_dept_result").show()
        // 对表进行缓存的方法
        spark.read.table("tmp_emp_join_dept_result").cache()
        spark.catalog.cacheTable("tmp_emp_join_dept_result")
        //输出到HDFS上
        // 方法一
        /*spark
                .read
                .table("tmp_emp_join_dept_result")
                .write.parquet("/spark/sql/hive_join_mysql")*/
        // 方法二
        spark
                .read
                .table("tmp_emp_join_dept_result")
                .write
                .format("parquet")
                .save(s"hdfs://hadoop01.com:8020/spark/sql/hive_join_mysql/${System.currentTimeMillis()}")
        //输出到Hive中，并且是parquet格式 按照deptno分区
        spark
                .read
                .table("tmp_emp_join_dept_result")
                .write
                .format("parquet")
                .partitionBy("deptno")
                .mode(SaveMode.Overwrite)
                .saveAsTable("hive_emp_dept")
        println("------------------------------------------------------------")
        spark.sql("show tables").show()
        //清空缓存
        spark.catalog.uncacheTable("tmp_emp_join_dept_result")
    }
}

bin/spark-submit \
--class com.jeffrey.sql.HiveJoinMySQLDemo \
--master yarn \
--deploy-mode client \
/opt/datas/jar/hivejoinmysql.jar
bin/spark-submit \
--class com.jeffrey.sql.HiveJoinMySQLDemo \
--master yarn \
--deploy-mode cluster \
/opt/datas/logAnalyze.jar