关于大数据spark的Hive集群到Hive集群的方案

object names {
    def main(args: Array[String]): Unit = {
        saveAsCsv()
        readCsvToHive()
    }

    //csv转成Hive表
    def readCsvToHive(): Unit = {
        val spark = SparkSession
                .builder()
                .appName("name1")
                .config("hive.metastore.uris", "目标集群")
                .enableHiveSupport()
                .getOrCreate()
        spark.sparkContext.setLogLevel("WARN")

        spark.read
                .options(Map("delimiter" -> "~", "header" -> "true"))
                .csv("/tmp/app_download_detail")
                .withColumn("create_time", from_unixtime(col("create_time"), "yyyy-MM-dd HH:mm:ss"))
                .withColumn("etl_created_time", from_unixtime(unix_timestamp(), "yyyy-MM-dd HH:mm:ss"))
                .write
                .mode("overwrite")
                .format("parquet")
                .saveAsTable("dwd.dwd_bury_app_download_detail_dd")

        spark.close()
    }

    //保存为CSV文件
    def saveAsCsv(): Unit = {
        val spark = SparkSession
                .builder()
                .master("local[4]")
                .appName("bury app download 01")
                .config("hive.metastore.uris", "源集群")
                .enableHiveSupport()
                .getOrCreate()
        spark.sparkContext.setLogLevel("WARN")
        import spark.implicits._

        val df = spark.read.table("表名")
                .repartition(4)
                .write
                .mode("overwrite")
                .format("csv")
                .option("sep", "~")
                .option("header", "true")
                .csv("/tmp/app_download_detail")

        spark.close()
    }
}