……
try {
    KuduTable kuduTable = kuduClient.openTable(tableName);
    KuduSession kuduSession = kuduClient.newSession();
    kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
    for(int i =0; i < dataArr.length; i++) {
        String dataStr = dataArr[i];
        Upsert upsert = kuduTable.newUpsert();
        PartialRow row = upsert.getRow();
        String[] dataInfo = dataStr.split(",");
        if(dataInfo.length == 4) {
            row.addInt("inv_item_sk", Integer.valueOf(dataInfo[0]));
            row.addInt("inv_warehouse_sk", Integer.valueOf(dataInfo[1]));
            row.addString("inv_date_sk", dataInfo[2]);
            row.addInt("inv_quantity_on_hand", Integer.valueOf(dataInfo[3]));
        }
        kuduSession.apply(upsert);
    }
    kuduSession.flush();
    kuduSession.close();
    System.out.println(" ******** KuduJavaSample -> upsert() successfule ! ");
} catch (KuduException e) {
    e.printStackTrace();
}
    ……

……
try {
  // Delete the table if it already exists.
  if(kuduContext.tableExists(tableName)) {
    kuduContext.deleteTable(tableName)
  }
……
  val write_rdd = spark.sparkContext.parallelize(write_arr)
  val write_df = spark.createDataFrame(write_arr.asJava, schema_df)
  kuduContext.insertRows(write_df, tableName)
  // Read from the table using an RDD.
  val read_cols = Seq("inv_item_sk", "inv_warehouse_sk", "inv_date_sk", "inv_quantity_on_hand")
  val rdd = kuduContext.kuduRDD(spark.sparkContext, tableName, read_cols)
  rdd.map { case Row(inv_item_sk: Int, inv_warehouse_sk: Int, inv_date_sk: String, inv_quantity_on_hand: Int) => (inv_item_sk, inv_warehouse_sk, inv_date_sk, inv_quantity_on_hand) }.
      collect().foreach(println(_))
  // Upsert some rows.
  val upsert_df = write_df.withColumn("inv_quantity_on_hand", col("inv_quantity_on_hand") + lit(1000))
  kuduContext.upsertRows(upsert_df, tableName)
  // Read the table in SparkSql.
  val read_df = spark.read.option("kudu.master", kuduMaster).
                option("kudu.table", tableName).
                format("kudu").load
  read_df.createOrReplaceTempView(tableName)
  spark.sqlContext.sql(s"select * from $tableName").show
……
} catch {
  case unknown : Throwable => log.error(s"got an exception: " + unknown)
……

……
val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", parmas.brokerList)
  .option("subscribe", parmas.sourceTopic)
  .option("startingOffsets", parmas.startPos)
  .option("failOnDataLoss", false)
  .load()
  .repartition(Integer.valueOf(parmas.partitionNum))
val schema = spark.read.format("hudi").load(s"${parmas.hudiBasePath}/${parmas.syncDB}/${parmas.syncTableName}").schema
val schema_bc = spark.sparkContext.broadcast(schema)
val tableType = if (parmas.hudiPartition != null && parmas.hudiPartition.length > 0) "MERGE_ON_READ" else "COPY_ON_WRITE"
val query = df.writeStream
  .queryName("MSK2Hudi")
  .foreachBatch { (batchDF: DataFrame, _: Long) =>
    if(batchDF != null && (!batchDF.isEmpty) ){
      val df = batchDF.withColumn("json", col("value").cast(StringType))
        .select(from_json(col("json"), schema_bc.value) as "data")
        .select("data.*")
        .withColumn("created_ts", lit((new Date()).getTime))
        .filter(genPrimaryKeyFilter(parmas.hudiKeyField))
      writeHudiTable(df, parmas.syncDB, parmas.syncTableName, "upsert", parmas.zookeeperUrl,
                      parmas.hudiKeyField, "created_ts", parmas.hudiPartition, parmas.hudiBasePath, tableType)
    }
  }
  .option("checkpointLocation", parmas.checkpointDir)
  .trigger(Trigger.ProcessingTime(parmas.trigger + " seconds"))
  .start
query.awaitTermination()
……
第3.部分，从ODS层读取数据写入DW层的测试代码如下：
……
//use item as dimension table
spark.read.format("hudi").
          load("s3://dalei-demo/hudi/kudu_migration/item").
          createOrReplaceTempView("item")
while(true){
  Thread.sleep(parmas.hudiIntervel)
  endTime = DATE_FORMAT.format(new Date())
  spark.read.format("hudi").
        option(QUERY_TYPE.key(), QUERY_TYPE_INCREMENTAL_OPT_VAL).
        option(BEGIN_INSTANTTIME.key(), beginTime).
        option(END_INSTANTTIME.key(), endTime).
        load("s3://dalei-demo/hudi/kudu_migration/inventory").
        createOrReplaceTempView("inventory")
  val df = spark.sql(
    s"""select in.inv_item_sk, nvl(i.i_brand, 'N/A') as i_brand, in.inv_warehouse_sk, in.inv_date_sk,
      |nvl(in.inv_quantity_on_hand, 0) as inv_quantity_on_hand,
      |${(new Date()).getTime} as created_ts
      |from inventory in left join item i on in.inv_item_sk = i.i_item_sk """.stripMargin)
  if(df.count > 0)
    writeHudiTable(df, parmas.syncDB, parmas.syncTableName, "upsert", parmas.zookeeperUrl,
        "inv_item_sk,i_brand,inv_warehouse_sk", "created_ts", "inv_date_sk", parmas.hudiBasePath, "MERGE_ON_READ")
  beginTime = endTime
}
……

……
// get all tables in the database to migrate.
val allTable = queryByJdbc(parmas.impalaJdbcUrl + parmas.kuduDatabase, "show tables")
if(allTable != null && allTable.isInstanceOf[Seq[String]] && allTable.length > 0) {
  allTable.foreach( tableName => {
    val (primaryKey, partitionKey) = getPrimaryAndPartitionKey(parmas.impalaJdbcUrl + parmas.kuduDatabase, tableName)
    val df = spark.read
      .option("kudu.master", parmas.kuduMaster)
      .option("kudu.table", "impala::" + parmas.kuduDatabase + "." + tableName)
      .format("kudu").load
      .filter(genPrimaryKeyFilter(primaryKey))
      .withColumn("created_ts", lit((new Date()).getTime))
      .repartition(parmas.partitionNum)
    val tableType = if (partitionKey != null && partitionKey.length > 0) "MERGE_ON_READ" else "COPY_ON_WRITE"
    writeHudiTable(df, parmas.syncDB, tableName, "bulk_insert", parmas.zookeeperUrl,
                      primaryKey, "created_ts", partitionKey, parmas.hudiBasePath, tableType)
  })
……

……
val df = spark.read
  .option("kudu.master", parmas.kuduMaster)
  .option("kudu.table", "impala::" + parmas.kuduDatabase + "." + parmas.syncTableName)
  .format("kudu").load
  .withColumn("created_ts", lit((new Date()).getTime))
  .filter(parmas.filterString)
  .select(to_json(struct("*"), Map("dropFieldIfAllNull" -> "false")).as("value"))
  .selectExpr(s"cast('${UUID.randomUUID().toString}' as string)", "cast(value as string)")
df.write
  .format("kafka")
  .option("kafka.bootstrap.servers", parmas.brokerList)
  .option("topic", parmas.sourceTopic)
  .save()
……

……
val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", parmas.brokerList)
      .option("subscribePattern", parmas.sourceTopic)
      .option("startingOffsets", parmas.startPos)
      .option("failOnDataLoss", false)
      .load()
      .repartition(parmas.partitionNum)
val query = df.writeStream
      .queryName("MSK2Hudi")
      .foreachBatch { (batchDF: DataFrame, _: Long) =>
        if(batchDF != null && (!batchDF.isEmpty) )
          writeMultiTable2HudiFromDF(batchDF, parmas.syncDB, "upsert", parmas.zookeeperUrl,
            parmas.hudiBasePath, parmas.impalaJdbcUrl, parmas.kuduDatabase)
      }
      .option("checkpointLocation", parmas.checkpointDir)
      .trigger(Trigger.ProcessingTime(parmas.trigger + " seconds"))
      .start
      ……