package main.scala.Mimport org.apache.spark.sql.{DataFrame, SparkSession}object amJobTask {def main(args: Array[String]): Unit = {// 1. 创建SparkSession并启用Hive支持val spark: SparkSession = SparkSession.builder().appName("job_name_id").master("local[10]")// .config("hive.metastore.uris", "thrift://hadoop01:9083") // 在 window的hosts配置IP映射：192.168.1.1 master//.config("spark.sql.warehouse.dir", "hdfs://master:9000/opt/hiveDataFile") //配置远程hdfs数据文件存储路径.enableHiveSupport().getOrCreate()// val df = spark.sql("select * from test_db.room2")// df.printSchema()// df.show()// 2. 配置MySQL连接参数val jdbcUrl = "jdbc:mysql://localhost:3306/hivedata?useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf-8"val mysqlProps = new java.util.Properties()mysqlProps.setProperty("user", "root")mysqlProps.setProperty("password", "123456")//mysqlProps.setProperty("driver", "com.mysql.jdbc.Driver")mysqlProps.setProperty("driver","com.mysql.cj.jdbc.Driver")// 3. 从MySQL读取数据
// val querSql:String = "(SELECT r.Gender,count(*) as cnt from room1 r group by r.Gender) as tmp"
// val mysqlDF: DataFrame = spark.read.jdbc(jdbcUrl, querSql, mysqlProps)
// mysqlDF.show()// 4。注册表在查询val tab_name: List[String] = List("room1","room2")// 遍历创建视图for (table <- tab_name) {val mysqlDF: DataFrame = spark.read.jdbc(jdbcUrl, table, mysqlProps)mysqlDF.createOrReplaceTempView(table)}// val mysqlDF: DataFrame = spark.read.jdbc(jdbcUrl, "room1", mysqlProps)
// mysqlDF.createOrReplaceTempView("room1")// val rls: DataFrame = spark.sql("SELECT r.Gender,count(*) as cnt from room1 r " +
// " where length(r.Gender)>0 and r.address like '%上海%' group by r.Gender")
// rls.show()val rls1: DataFrame = spark.sql("SELECT r.Gender,count(*) as cnt from room1 r group by r.Gender \nunion all\nSELECT r.Gender,count(*) as cnt from room2 r group by r.Gender ")rls1.show(100)// 获取所有临时视图（会话级别）val tempViews = spark.catalog.listTables().filter(_.isTemporary).select("name").collect().map(_.getString(0))tempViews.foreach(println)// 获取全局临时视图
// val globalTempViews = spark.catalog.listTables("global_temp")
// .select("name")
// .collect()
// .map(_.getString(0))
//
// println("Global temporary views:")
// globalTempViews.foreach(println)spark.sql("select COUNT(1) CNT from (\nselect * from room1 where 1>0\nunion all\nselect * from room2 where 20>3\nunion all\nselect * from room1 where 10>5\nunion all\nselect * from room2 where 223>5\nunion all\nselect * from room1 where 12>9\nunion all\nselect * from room2 where 4>1.5\nunion all\nselect * from room1 where 9>8\nunion all\nselect * from room2 where 7>6\nunion all\nselect * from room1 where 4>2\nunion all\nselect * from room2 where 7>6\n)").show()// spark.sql("select count(*) as cnt from room1\nunion all\nselect count(*) as cnt from room2").show()val datacnt:DataFrame = spark.sql("select T.Gender,COUNT(1) as CNT from (\nselect * from room1\nunion all\nselect * from room2\nunion all\nselect * from room1\nunion all\nselect * from room2\nunion all\nselect * from room1\nunion all\nselect * from room2\nunion all\nselect * from room1\nunion all\nselect * from room2\n) t where t.Gender in('M','F')\n\tand SUBSTR(T.Version,1,10)>= '1910-01-01'\ngroup by T.Gender")datacnt.show()datacnt.write.mode("append").jdbc(jdbcUrl,"room_cnt",mysqlProps)spark.stop()}}
