Skip to content
Permalink
Browse files
[HUDI-2225] Add a compaction job in hudi-examples (#3347)
  • Loading branch information
codope committed Aug 3, 2021
1 parent b21ae68 commit aa857beee00a764cee90d6e790ee4b0ab4ad4862
Showing 1 changed file with 113 additions and 0 deletions.
@@ -0,0 +1,113 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/

package org.apache.hudi.examples.spark

import org.apache.hudi.DataSourceWriteOptions.{PARTITIONPATH_FIELD_OPT_KEY, PRECOMBINE_FIELD_OPT_KEY, RECORDKEY_FIELD_OPT_KEY, TABLE_TYPE_OPT_KEY}
import org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs
import org.apache.hudi.client.SparkRDDWriteClient
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.common.model.{HoodieAvroPayload, HoodieRecordPayload, HoodieTableType}
import org.apache.hudi.common.util.Option
import org.apache.hudi.config.HoodieWriteConfig.TABLE_NAME
import org.apache.hudi.config.{HoodieCompactionConfig, HoodieWriteConfig}
import org.apache.hudi.examples.common.{HoodieExampleDataGenerator, HoodieExampleSparkUtils}
import org.apache.spark.sql.SaveMode.{Append, Overwrite}
import org.apache.spark.sql.SparkSession

import scala.collection.JavaConverters._

/**
* Simple example to run a compaction job for MOR table.
* To run this example, you should:
* 1. For running in IDE, set VM options `-Dspark.master=local[2]`
* 2. For running in shell, using `spark-submit`
*
* Usage: HoodieMorCompactionJob <tablePath> <tableName>.
* <tablePath> and <tableName> describe root path of hudi and table name
* for example, `HoodieMorCompactionJob file:///tmp/hoodie/hudi_mor_table hudi_mor_table`
*/
object HoodieMorCompactionJob {

def main(args: Array[String]): Unit = {
if (args.length < 2) {
System.err.println("Usage: HoodieMorCompactionJob <tablePath> <tableName>")
System.exit(1)
}

val spark = HoodieExampleSparkUtils.defaultSparkSession("Hudi MOR table compaction via Spark example")
val dataGen = new HoodieExampleDataGenerator[HoodieAvroPayload]
val tablePath = args(0)
val tableName = args(1)
insertData(spark, tablePath, tableName, dataGen, HoodieTableType.MERGE_ON_READ.name())
updateData(spark, tablePath, tableName, dataGen, HoodieTableType.MERGE_ON_READ.name())
val cfg = HoodieWriteConfig.newBuilder()
.withPath(tablePath)
.withSchema(HoodieExampleDataGenerator.TRIP_EXAMPLE_SCHEMA)
.withParallelism(2, 2)
.forTable(tableName)
.withCompactionConfig(HoodieCompactionConfig.newBuilder()
.withInlineCompaction(true)
.withMaxNumDeltaCommitsBeforeCompaction(1).build())
.build()
val client = new SparkRDDWriteClient[HoodieRecordPayload[Nothing]](new HoodieSparkEngineContext(spark.sparkContext), cfg)
try {
val instant = client.scheduleCompaction(Option.empty())
client.compact(instant.get())
client.clean()
} catch {
case e: Exception => System.err.println(s"Compaction failed due to", e)
} finally {
client.close()
spark.stop()
}
}

def insertData(spark: SparkSession, tablePath: String, tableName: String,
dataGen: HoodieExampleDataGenerator[HoodieAvroPayload], tableType: String): Unit = {
val commitTime: String = System.currentTimeMillis().toString
val inserts = dataGen.convertToStringList(dataGen.generateInserts(commitTime, 20))
val df = spark.read.json(spark.sparkContext.parallelize(inserts.asScala, 1))
df.write.format("org.apache.hudi").
options(getQuickstartWriteConfigs).
option(PRECOMBINE_FIELD_OPT_KEY.key, "ts").
option(RECORDKEY_FIELD_OPT_KEY.key, "uuid").
option(PARTITIONPATH_FIELD_OPT_KEY.key, "partitionpath").
option(TABLE_NAME.key, tableName).
option(TABLE_TYPE_OPT_KEY.key, tableType).
mode(Overwrite).
save(tablePath)
}

def updateData(spark: SparkSession, tablePath: String, tableName: String,
dataGen: HoodieExampleDataGenerator[HoodieAvroPayload], tableType: String): Unit = {
val commitTime: String = System.currentTimeMillis().toString
val updates = dataGen.convertToStringList(dataGen.generateUpdates(commitTime, 10))
val df = spark.read.json(spark.sparkContext.parallelize(updates.asScala, 1))
df.write.format("org.apache.hudi").
options(getQuickstartWriteConfigs).
option(PRECOMBINE_FIELD_OPT_KEY.key, "ts").
option(RECORDKEY_FIELD_OPT_KEY.key, "uuid").
option(PARTITIONPATH_FIELD_OPT_KEY.key, "partitionpath").
option(TABLE_NAME.key, tableName).
option(TABLE_TYPE_OPT_KEY.key, tableType).
mode(Append).
save(tablePath)
}
}

0 comments on commit aa857be

Please sign in to comment.