spark-shell --conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" --conf "spark.sql.hive.convertMetastoreParquet=false" --jars /usr/lib/hudi/hudi-spark-bundle.jar,/usr/lib/spark/external/lib/spark-avro.jar

val ggDeltaFiles = "s3://" + sourceBucket + "/" + sourceSubFolder + "/" + sourceSystem + "/" + inputTableName + "/";
val rootDataframe:DataFrame = spark.read.format("avro").load(ggDeltaFiles);

// extract PK fields name from first line
val pkFields: Seq[String] = rootDataframe.select("primary_keys").limit(1).collect()(0).getSeq(0);

// take into account the "after." fields only
val columnsPre:Array[String] = rootDataframe.select("after.*").columns;

// exclude "_isMissing" fields added by Oracle GoldenGate
// The second part of the expression will safely preserve all native "**_isMissing" fields
val columnsPost:Array[String] = columnsPre.filter { x => (!x.endsWith("_isMissing")) || (!x.endsWith("_isMissing_isMissing") && (columnsPre.filter(y => (y.equals(x + "_isMissing")) ).nonEmpty))};
val columnsFinal:ArrayBuffer[String] = new ArrayBuffer[String]();

columnsFinal += "op_ts";
columnsFinal += "pos";

// add the "after." prefix
columnsPost.foreach(x => (columnsFinal += "after." + x));

// prepare the target dataframe with the partition additional column
val preparedDataframe = rootDataframe.select("opTypeFieldName", columnsFinal.toArray:_*).
  withColumn("HUDI_PART_DATE", date_format(to_date(col("DAT_ORDER"), "yyyy-MM-dd"),"yyyy/MM/dd")).
  filter(col(opTypeFieldName).isin(admittedValues.toList: _*));

// write data
preparedDataframe.write.format("org.apache.hudi").
  options(hudiOptions).
  option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, pkFields.mkString(",")).
  mode(SaveMode.Append).
  save(hudiTablePath);