scala代码高速拉去数据-适合同步历史数据场景

package com.longi.util

import com.longi.common.OracleTemplate
import com.longi.common.base.{HDFSPathWrapper, PropertiesHelper}
import com.longi.hadoop.{LoadKuduToDataFrame, SparkSessionPort}
import org.apache.kudu.client.SessionConfiguration
import org.apache.kudu.spark.kudu.{KuduContext, KuduWriteOptions}
import org.apache.spark.sql.functions.{coalesce, lit}
import org.apache.spark.sql.types.{DataType, DecimalType, StructField, StructType}
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

/**
 * @date: 2021-12-3 9:55
 * @desc: Please fill in the remarks
 */
object Oracle2KuduWMS {

  def main(args: Array[String]): Unit = {
    //Step 0: 参数初始化
    if (args.length > 7) {
      throw new IllegalArgumentException("You need to pass the following parameters  1:The data of time(2020-06-01) 2:The name of Kudu table 3:Data extraction date column 4:The business Domain")
    }

    //(Oracle Table , Oracle Column, Begin Dt, End Dt, interval, Kudu Table)
    //Oracle Table Name
    val oraTableName = args(0)
    //Oracle Incremental Sliding Column
    val oraColumnName = args(1)
    //The Begin time and End Time
    val datBeginDateTime = args(2)
    val datEndDateTime = args(3)
    //Kudu table name
    val kuduTableName = args(4)
    val mesge = s"Oracle Table $oraTableName -> Kudu Table : $kuduTableName ($datBeginDateTime - $datEndDateTime) "
    val fieldValueChange = args(5)
    println(mesge)
    val spark = SparkSessionPort(mesge)

    var exeSQLCondition = ""
    if (datBeginDateTime == "0000-00-00 00:00:00") {
      exeSQLCondition = "1=1"
    } else {
      exeSQLCondition = oraColumnName.concat(">=to_date('").concat(datBeginDateTime).concat("','yyyy-mm-dd hh24:mi:ss') AND ").concat(oraColumnName).concat("<to_date('").concat(datEndDateTime).concat("','yyyy-mm-dd hh24:mi:ss')")
    }
    println(s"exeSQLCondition = ${exeSQLCondition}")

    val exeSQL =
      s"""
         |     (
         |     SELECT rownum as rs
         |           ,rowid as offsets  -- 用这个替换cast(rowid as varchar(20)) as offsets
         |           ,'I' as op_type
         |           ,t.*
         |       FROM ${oraTableName} t
         |      WHERE ${exeSQLCondition}
         |     )
          """.stripMargin
    println(s"exeSQL = ${exeSQL}")
    val exeAggrSQL =
      s"""
         |     (
         |     SELECT count(1)
         |       FROM ${oraTableName} t
         |      WHERE ${exeSQLCondition}
         |     ) t
         |""".stripMargin
    println(exeAggrSQL)

    val aggrDF = spark
      .read
      .format("jdbc")
      .option("url", PropertiesHelper.getPropertiesValueFromKey("origi.wms.ora.uri"))
      .option("driver", "oracle.jdbc.driver.OracleDriver")
      .option("dbtable", exeAggrSQL)
      .option("user", PropertiesHelper.getPropertiesValueFromKey("origi.wms.ora.username"))
      .option("password", PropertiesHelper.getPropertiesValueFromKey("origi.wms.ora.password")).load()
    val upperBound = aggrDF.first().getDecimal(0).toBigInteger.intValue()


    val numPartitions = upperBound match {
      case upperBound if (upperBound <= 100000) => 1
      case upperBound if (upperBound < 100000 && upperBound <= 500000) => 3
      case upperBound if (upperBound < 500000 && upperBound <= 1000000) => 10
      case upperBound if (upperBound < 1000000 && upperBound <= 5000000) => 15
      case _ => 20
    }

    val oracleDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:oracle:thin:@xxx-st.longi.com:1521/prod")
      .option("driver", "oracle.jdbc.driver.OracleDriver")
      .option("dbtable", exeSQL)
      .option("user", "USER")
      .option("partitionColumn", "rs")
      .option("lowerBound", 1)
      .option("upperBound", upperBound)
      .option("numPartitions", numPartitions)
      .option("fetchsize", 3000)
      .option("password", "XXXXXXX").load()
    println(oracleDF.printSchema())

    val kuduSchema = LoadKuduToDataFrame(spark, kuduTableName).schema
    import org.apache.spark.sql.functions._
    //按照Kudu表数据类型对Oracle数据类型进行改造
    val colNames = kuduSchema.names.map(c => {
      col(c).cast(kuduSchema.fields(kuduSchema.fieldIndex(c)).dataType.typeName)
    })
    var oracleFinalDF = oracleDF.select(colNames: _*)
    //oracleFinalDF.show()
    //需要转变的字段不为空
    /*val fields = fieldValueChange.split(",")

    fields.foreach(f => {
      /*oracleFinalDF.schema.fields(kuduSchema.fieldIndex(f.split("=")(0))).dataType.typeName match {
        case "Int" => coalesce(oracleFinalDF(f.split("=")(0)), lit(f.split("=")(1)))
        case "String" => (f.split("=")(0))
      }*/
      println(s"f={$fields}, f0=${f.split("=")(0)} f1=${f.split("=")(1)}")
      //coalesce(oracleFinalDF(f.split("=")(0)), lit(f.split("=")(1)))
      //nvl(oracleFinalDF.col(f.split("=")(0)), f.split("=")(1))
      //coalesce(col(f.split("=")(0)),lit(oracleFinalDF(f.split("=")(1))))
      val tf = oracleFinalDF.withColumn(f.split("=")(0).concat("_new"), nvl(oracleFinalDF.col(f.split("=")(0)), f.split("=")(1)))
      //tf.withColumn("a",oracleFinalDF.col("gl_sl_link_id"))
      tf.where("").show()
      val df = tf.drop(col(f.split("=")(0)))
      df.show()
      val rf = df.withColumnRenamed(f.split("=")(0).concat("_new"), f.split("=")(0))
      rf.where("gl_sl_link_id is null or gl_sl_link_id =-1").show()
      oracleFinalDF = rf
    })

    //将空值转为给定值
    def nvl(ColIn: Column, ReplaceVal: Any): Column = {
      println(ReplaceVal)
      (when(ColIn.isNull, lit(ReplaceVal)).otherwise(ColIn))
    }

    oracleFinalDF.select("gl_sl_link_id").where("gl_sl_link_id is null or gl_sl_link_id =-1").distinct().show(200)
*/
    //创建新DF，用于插入Kudu表中
    val newData = spark.createDataFrame(oracleFinalDF.rdd, kuduSchema)


    println(newData.printSchema())

    println(s"newData.rdd.partitions.size :${newData.rdd.partitions.size}")

    upsertKuduData(spark, newData, kuduTableName)
    spark.stop()

  }

  /**
   * @Description:写入Kudu数据库
   * @Param: [spark, dataFrame, tabName]
   * @return: void
   */
  private def upsertKuduData(spark: SparkSession, dataFrame: DataFrame, tabName: String): Unit = {
    val kuduContext = new KuduContext(
      HDFSPathWrapper.getKuduMasterPath(),
      spark.sparkContext
    )
    val ks = kuduContext.syncClient.newSession()
    ks.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC)
    ks.setMutationBufferSpace(10000)
    kuduContext.upsertRows(dataFrame, tabName, new KuduWriteOptions(false, true))
  }

}

调用方式

spark-submit  --class com.longi.util.Oracle2KuduWMS  --master yarn  --deploy-mode cluster  --queue bdp --driver-memory 2g  --num-executors 20  --executor-memory 1g  --executor-cores 1  --conf spark.sql.warehouse.dir=hdfs://longi/home/data/hive/warehouse  --conf spark.sql.session.timeZone=Asia/Shanghai  --conf spark.executor.memoryOverhead=10G  --jars /dfs/projects/etl-schedule-entry/lib/common/etl-functions-1.0.jar,/dfs/projects/etl-schedule-entry/lib/common/ojdbc6-11.2.0.3.jar,/dfs/projects/etl-schedule-entry/lib/common/kudu-spark2_2.11-1.10.0-cdh6.3.2.jar,/dfs/projects/etl-schedule-entry/lib/common/grizzled-slf4j_2.11-1.3.0.jar,/dfs/projects/etl-schedule-entry/lib/common/mysql-connector-java-5.1.44.jar  /dfs/projects-test/etl-schedule-entry/lib/common/etl-tools-1.0.jar WMS_PROD.ACT_ALLOCATION_DETAILS  addtime '2010-01-01 00:00:00' '2023-12-23 10:00:00'  ods_po_wms.streaming_cemd_act_allocation_details  'a=1'