zl程序教程

您现在的位置是:首页 >  数据库

当前栏目

Spark读取Hive数据的两种方式与保存数据到HDFS

2023-04-18 16:26:39 时间

Spark读取Hive数据的两种方式与保存数据到HDFS

Spark读取Hive数据的方式主要有两种

1、 通过访问hive metastore的方式,这种方式通过访问hive的metastore元数据的方式获取表结构信息和该表数据所存放的HDFS路径,这种方式的特点是效率高、数据吞吐量大、使用spark操作起来更加友好。

2、 通过spark jdbc的方式访问,就是通过链接hiveserver2的方式获取数据,这种方式底层上跟spark链接其他rdbms上一样,可以采用sql的方式先在其数据库中查询出来结果再获取其结果数据,这样大部分数据计算的压力就放在了数据库上。

两种方式的具体实现示例

首先创建Spark Session对象:

    val spark = SparkSession.builder()
      .appName("test")
      .enableHiveSupport()
      .getOrCreate()

方式一(推荐) 直接采用Spark on Hive的方式读取数据,这样SparkSession在使用sql的时候会去找集群hive中的库表,加载其hdfs数据与其元数据组成DataFrame

val df = spark.sql("select * from test.user_info")

方式二 采用spark jdbc的方式,如果有特别的使用场景的话也可以通过这种方法来实现。

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.jdbc.{JdbcDialect, JdbcDialects}
 
 
object test{
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[2]")
      .appName("test")
      .getOrCreate()
 
    register() //如果不手动注册,只能获取到数据库中的表结构,而不能获取到数据
    val df = spark.read
      .format("jdbc")
      .option("driver","org.apache.hive.jdbc.HiveDriver")
      .option("url","jdbc:hive2://xxx:10000/")
      .option("user","hive")
      .option("password",xxx)
      .option("fetchsize", "2000")
      .option("dbtable","test.user_info")
      .load()
    df.show(10)
  }
 
  def register(): Unit = {
    JdbcDialects.registerDialect(HiveSqlDialect)
  }
 
 
  case object HiveSqlDialect extends JdbcDialect {
    override def canHandle(url: String): Boolean = url.startsWith("jdbc:hive2")
 
    override def quoteIdentifier(colName: String): String = {
      colName.split('.').map(part => s"`$part`").mkString(".")
    }
  }
 
}

Spark的DataFrame和DataSet使用

​ DataFrame是Spark SQL提供的一个编程抽象,与RDD类似,也是一个分布式的数据集合。但与RDD不同的是,DataFrame的数据都被组织到有名字的列中,就像关系型数据库中的表一样。此外,多种数据都可以转化为DataFrame,例如Spark计算过程中生成的RDD、结构化数据文件、Hive中的表、外部数据库等。

在Spark中,一个DataFrame所代表的是一个元素类型为Row的Dataset,即DataFrame只是Dataset[Row]的一个类型别名。相对于RDD,Dataset提供了强类型支持,在RDD的每行数据加了类型约束。而且使用DatasetAPI同样会经过Spark SQL优化器的优化,从而提高程序执行效率。

DataFrame和R的数据结构以及python pandas DataFrame的数据结构和操作基本一致。

创建DataFrame、DataSet

  • 创建RDD
  • RDD转化为ROW
  • 通过ROW和元数据信息生成DataFrame
  • 然后通过DataFrame和对应的类转化为DataSet
  • 也就是说DataFrame是DataSet[Row],这里可以通过指定的类将其转化,DataSet[User]
  • 需要注意的事转化使用的类需要时内部类,然后就是类里的变量名要和元数据信息的列名保持对齐。
object MovieLenDataSet {
  case class User(UserID:String, Gender:String, Age:String, Occupation:String, Zip_Code:String)
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val spark = SparkSession.builder()
      .appName("MovieLenDataSet")
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._

    val dataPath = "/home/ffzs/data/ml-1m"
    val schema4users = StructType(
      "UserID::Gender::Age::Occupation::Zip_code"
        .split("::")
        .map(it => StructField(it, StringType, nullable = true))
    )

    val usersRdd = spark.sparkContext.textFile(f"$dataPath/users.dat")
    val usersRows = usersRdd.map(_.split("::"))
      .map(it => {
        it.map(_.trim)
      })
      .map(it => Row(it(0), it(1), it(2), it(3), it(4)))
    val usersDF: DataFrame = spark.createDataFrame(usersRows, schema4users)
    val usersDataSet = usersDF.as[User]
    usersDataSet.show(5)
  }
}

Spark的DataFrame存储的Mode模式选择

spark的dataframe存储中都会调用write的mode方法:

data.write.mode(“append”).saveAsTable(s"u s e r i d . {userid}.userid.{datasetid}")
data.write.mode(SaveMode.Overwrite).parquet(hdfspath)

但不同时候的参数是不同的。

先看一下源码:

spark-v2.3.0:

  def mode(saveMode: SaveMode): DataFrameWriter[T] = {
    this.mode = saveMode
    this
  }

  /**
   * Specifies the behavior when data or table already exists. Options include:
   *   - `overwrite`: overwrite the existing data.
   *   - `append`: append the data.
   *   - `ignore`: ignore the operation (i.e. no-op).
   *   - `error` or `errorifexists`: default option, throw an exception at runtime.
   *
   * @since 1.4.0
   */
  def mode(saveMode: String): DataFrameWriter[T] = {
    this.mode = saveMode.toLowerCase(Locale.ROOT) match {
      case "overwrite" => SaveMode.Overwrite
      case "append" => SaveMode.Append
      case "ignore" => SaveMode.Ignore
      case "error" | "errorifexists" | "default" => SaveMode.ErrorIfExists
      case _ => throw new IllegalArgumentException(s"Unknown save mode: $saveMode. " +
        "Accepted save modes are 'overwrite', 'append', 'ignore', 'error', 'errorifexists'.")
    }
    this
  }

SaveMode.Overwrite(对应着字符串"overwrite"):表示如果目标文件目录中数据已经存在了,则用需要保存的数据覆盖掉已经存在的数据

SaveMode.Append(对应着字符串"append"):表示如果目标文件目录中数据已经存在了,则将数据追加到目标文件中
数据追加方式是:先将表中的所有索引删除,再追加数据

SaveMode.Ignore(对应着字符串为:“ignore”):表示如果目标文件目录中数据已经存在了,则不做任何操作
SaveMode.ErrorIfExists(对应着字符串"error"):表示如果目标文件目录中数据已经存在了,则抛异常(这个是默认的配置)


spark之Dataframe保存模式

以前spark.write时总要先把原来的删了,但其实是可以设置写入模式的。

val df =  spark.read.parquet(input)
df.write.mode("overwrite").parquet(output)

dataframe写入的模式一共有4种:

  1. overwrite 覆盖已经存在的文件
  2. append 向存在的文件追加
  3. ignore 如果文件已存在,则忽略保存操作
  4. error / default 如果文件存在,则报错
def mode(saveMode: String): DataFrameWriter = {
    this.mode = saveMode.toLowerCase match {
      case "overwrite" => SaveMode.Overwrite              
      case "append" => SaveMode.Append                    
      case "ignore" => SaveMode.Ignore                    
      case "error" | "default" => SaveMode.ErrorIfExists  
      case _ => throw new IllegalArgumentException(s"Unknown save mode: $saveMode. " +
        "Accepted modes are 'overwrite', 'append', 'ignore', 'error'.")
    }
    this
  }

spark write写入数据task failed失败,两种模式下的不同表现

1、SaveMode.Append

task失败重试,并不会删除上一次失败前写入的数据(文件根据分区号命名),重新执行时会继续追加数据。所以会出现数据重复。

2、SaveMode.Overwrite

task失败重试,会删除该分区上次失败所写入的数据文件,然后创建一个新的数据文件写入数据。所以不会出现数据重复。

启动spark任务报错:ERROR SparkUI: Failed to bind SparkUI

在这里插入图片描述
当启动一个spark任务的时候,就会占用一个端口,默认为4040,从日志可以看到当端口被占用时,它会默认依次增加16次到4056,如果还是失败的话,就会报错退出。

解决方法:

  1. 使用spark-submit提交任务时,在脚本中加配置:–conf spark.port.maxRetries=128(亲测有效)

以下代码仅供学习参考

Spark执行外部Hql脚本Scala代码Demo示例

  • pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>SparkReadHql_Test</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <maven.compiler.encoding>UTF-8</maven.compiler.encoding>
        <encoding>UTF-8</encoding>
        <hadoop.version>3.1.3</hadoop.version>
        <hive.version>3.1.2</hive.version>
        <scala.version>2.12.11</scala.version>
        <spark.version>3.0.0</spark.version>
    </properties>

    <dependencies>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
            <!--将netty包排除-->
            <exclusions>
                <exclusion>
                    <groupId>io.netty</groupId>
                    <artifactId>netty</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <!--解决io.netty.buffer.PooledByteBufAllocator.defaultNumHeapArena()I异常,-->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.18.Final</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>${hadoop.version}</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.scala-lang/scala-library -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>${spark.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>com.google.guava</groupId>
                    <artifactId>guava</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-hive -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <!--      解决问题引的包: org.apache.hadoop.hive.hbase.hbaseserde not found-->


        <!--        <dependency>-->
        <!--            <groupId>org.apache.hadoop</groupId>-->
        <!--            <artifactId>hadoop-hdfs</artifactId>-->
        <!--            <version>${hadoop.version}</version>-->
        <!--        </dependency>-->

        <!--        <dependency>-->
        <!--            <groupId>org.apache.hadoop</groupId>-->
        <!--            <artifactId>hadoop-hdfs-client</artifactId>-->
        <!--            <version>${hadoop.version}</version>-->
        <!--        </dependency>-->

        <!--        <dependency>-->
        <!--            <groupId>org.apache.hbase</groupId>-->
        <!--            <artifactId>hbase-client</artifactId>-->
        <!--            <version>${hbase.version}</version>-->
        <!--        </dependency>-->
        <!--        <dependency>-->
        <!--            <groupId>org.apache.hbase</groupId>-->
        <!--            <artifactId>hbase-common</artifactId>-->
        <!--            <version>${hbase.version}</version>-->
        <!--        </dependency>-->
        <!--        <dependency>-->
        <!--            <groupId>org.apache.hbase</groupId>-->
        <!--            <artifactId>hbase-http</artifactId>-->
        <!--            <version>${hbase.version}</version>-->
        <!--        </dependency>-->
        <!--spark-hive begin-->
        <!--        <dependency>-->
        <!--            <groupId>org.apache.hive</groupId>-->
        <!--            <artifactId>hive-serde</artifactId>-->
        <!--            <version>${hive.version}</version>-->
        <!--        </dependency>-->

        <!--        <dependency>-->
        <!--            <groupId>org.apache.hive</groupId>-->
        <!--            <artifactId>hive-exec</artifactId>-->
        <!--            <version>${hive.version}</version>-->
        <!--            <exclusions>-->
        <!--                <exclusion>-->
        <!--                    <groupId>org.apache.avro</groupId>-->
        <!--                    <artifactId>avro</artifactId>-->
        <!--                </exclusion>-->
        <!--            </exclusions>-->
        <!--        </dependency>-->

        <!--        <dependency>-->
        <!--             <groupId>org.apache.hive.hcatalog</groupId>-->
        <!--             <artifactId>hive-hcatalog-core</artifactId>-->
        <!--             <version>${hive.version}</version>-->
        <!--             <exclusions>-->
        <!--             <exclusion>-->
        <!--             <groupId>org.apache.avro</groupId>-->
        <!--             <artifactId>avro</artifactId>-->
        <!--             </exclusion>-->
        <!--             </exclusions>-->
        <!--        </dependency>-->
        <!--spark-hive end-->
        <!-- https://mvnrepository.com/artifact/com.google.guava/guava -->
        <!--        <dependency>-->
        <!--            <groupId>com.google.guava</groupId>-->
        <!--            <artifactId>guava</artifactId>-->
        <!--            <version>15.0</version>-->
        <!--        </dependency>-->
        <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
        <!--        <dependency>-->
        <!--            <groupId>mysql</groupId>-->
        <!--            <artifactId>mysql-connector-java</artifactId>-->
        <!--            <version>5.1.46</version>-->
        <!--        </dependency>-->
        <!-- https://mvnrepository.com/artifact/org.apache.hive/hive-cli -->
        <!--        <dependency>-->
        <!--            <groupId>org.apache.hive</groupId>-->
        <!--            <artifactId>hive-cli</artifactId>-->
        <!--            <version>${hive.version}</version>-->
        <!--        </dependency>-->

        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.15</version>
            <exclusions>
                <exclusion>
                    <groupId>javax.jms</groupId>
                    <artifactId>jms</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jdmk</groupId>
                    <artifactId>jmxtools</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jmx</groupId>
                    <artifactId>jmxri</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

    </dependencies>

    <build>
        <sourceDirectory>src/main/java</sourceDirectory>

        <plugins>

            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>

        </plugins>
    </build>


</project>
  • 代码
package com.xxxx


import org.apache.spark.sql.SparkSession
import java.io.File
import java.io.FileInputStream


import scala.io.{BufferedSource, Source}


object SparkReadHqlTest {

  def main(args: Array[String]): Unit = {

    val filePath: String = args(0)
    val input_date: String = args(1)

    val session: SparkSession = SparkSession.builder()//.master("local[2]")
      .appName("SparkSeesionApp")
      .enableHiveSupport() //支持hive
      .getOrCreate()
    //    session.sparkContext.setLogLevel("WARN")

    val sql: String = doFile(filePath)
    val strings: Array[String] = sql.split(";")
    var i = 0;
    strings.foreach(sql=>{
      val startTime: Long = System.currentTimeMillis()
      println("==============第 "+(i+1)+" 次===sql开始=================")
      println(sql)
      //替换参数
     // session.sql(sql.replace("'${hivevar:input_date}'", input_date)).show()
      session.sql(sql).show()
      val stopTime: Long = System.currentTimeMillis()
      val processTime: Long = (startTime - stopTime) / 1000
      println("===============第 "+(i+1)+" 次==sql结束====耗时=="+processTime+" 秒==========")
      i = i+1
    })
    //关闭SparkSession
    session.stop()
  }

  //读取外部sql文件文件
  def doFile(fileName: String): String = {
    val file: File = new File(fileName)

    val stream: FileInputStream = new FileInputStream(file)
    val buff: BufferedSource = Source.fromInputStream(stream,"UTF-8")
    //读取拼装SQL
    val sql: String = buff.getLines().mkString("
")
    sql
  }
}

Spark读取外部SQL文件java代码

  • pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>SparkReadHqlFile</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.1.1</version>
        </dependency>
        <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>3.1.1</version>
    </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>3.2.1</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.15</version>
            <exclusions>
                <exclusion>
                    <groupId>javax.jms</groupId>
                    <artifactId>jms</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jdmk</groupId>
                    <artifactId>jmxtools</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jmx</groupId>
                    <artifactId>jmxri</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

    </dependencies>

</project>
  • 代码
package org.example;

import org.apache.commons.lang3.StringUtils;
import  org.slf4j.Logger;
import org.apache.spark.sql.SparkSession;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class SparkReadFile {

private static final Logger logger= LoggerFactory.getLogger(SparkReadFile.class);

    public static void main(String[] args) throws IOException {
     // 传入参数非空判断
    validateArgs(args);
   // 1.创建sparkSession
        SparkSession spark = SparkSession.builder().config("hive.metastore.uris", args[1]) //hive的metastore地址
                .config("Spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .config("hive.exec.dynamic.partition.mode", "nonstrict").enableHiveSupport().getOrCreate();
        // 2.解析sql文件
        BufferedReader bufferedReader =null;
        String tmpStr;
        String execStatus ="";
        try {
            bufferedReader =new BufferedReader(new FileReader(args[0])); //sql文件名
            StringBuilder tempSqlContent =new StringBuilder();
           while((tmpStr =bufferedReader.readLine()) !=null){
               tempSqlContent.append(tmpStr+"
");
           }
        // 替代sql语句中的变量${batchDate}为对应的分区信息
            String[] sqlList = tempSqlContent.toString().replaceAll("\$\{batchDate\}", args[2]).split(";");
            for (int i = 0; i <sqlList.length-1; i++) {
                logger.info("sql语句:{}",sqlList[i]);
              // 3.执行SQL语句
              spark.sql(sqlList[i]).show(false);
            }
        } catch (Exception e) {
           logger.error("
作业执行失败,{}
"+e.getMessage(),e);
           execStatus="1";
        } finally {
               // 4.关闭流
               if (null !=bufferedReader){
                   bufferedReader.close();
               }
               if (null !=spark){
                   spark.close();
               }
               if ("1".equals(execStatus)){
                   System.exit(-1);
               }
        }
    }
 //参数非空判断
    public static void validateArgs(String[] agrs){
        if (null == agrs || args.length !=3 || StringUtils.isAnyEmpty(args)){
            System.exit(-1);
        }
    }
}

租户Spark跨集群读取hive数据进行保存

  • pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>MasterClusterToZH</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>3.2.1</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.15</version>
            <exclusions>
                <exclusion>
                    <groupId>javax.jms</groupId>
                    <artifactId>jms</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jdmk</groupId>
                    <artifactId>jmxtools</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jmx</groupId>
                    <artifactId>jmxri</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

    </dependencies>
</project>
  • 代码
package org.example;


import org.apache.commons.lang3.StringUtils;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ReadHiveToTenant {
    private static final Logger logger= LoggerFactory.getLogger(ReadHiveToTenant.class);

    public static void main(String[] args) {
   // 传入参数非空判断
    validateArgs(args);
  // 1.初始化SparkSession对象,主集群Metastore
        SparkSession spark = createSparkSession(args[0]);
   // 2.读取主集群hive表分区数据
        Dataset<Row> partitonDF = spark.read().table(args[1]).where(args[2]);
        long count =partitonDF.count();
        logger.info(args[1]+"表的数据量:-----:"+count);
    //3.写入租户hive表的HDFS路径
      partitonDF.write().mode(SaveMode.Overwrite).save(args[3]);
    //4.关闭资源
    if (null !=spark){
        spark.close();
     }
    //5.修复hive表
      SparkSession tenantSparkSession = createSparkSession(args[4]) ;
        tenantSparkSession.sql("MSCK REPAIR TABLE"+agrs[5]);
        if (null !=tenantSparkSession){
            tenantSparkSession.close();
        }
    }
  // 获取一个SparkSession对象
 public static SparkSession createSparkSession(String hiveMetastore){
     SparkSession sparkSession = SparkSession.builder().config("hive.metastore.uris", hiveMetastore)
             .config("hive.exec.dynamic.partition", true) //开启动态分区
             .config("hive.exec.dynamic.partition.mode", "nonstrict").enableHiveSupport().getOrCreate();
     return sparkSession;

 }


    //参数非空判断
    public static void validateArgs(String[] agrs){
        if (null == agrs || args.length !=6 || StringUtils.isAnyEmpty(args)){
            System.exit(-1);
        }
    }
}

以上代码示例仅供学习参考,方便收藏,对代码进行整理汇总
参考博客:

  • https://blog.csdn.net/qq_42213403/article/details/117557610
  • https://blog.csdn.net/qq_34009542/article/details/118366474?spm=1001.2014.3001.5502
  • https://blog.csdn.net/totally123/article/details/117224169