E-MapReduce:Basic usage

Prerequisites

Before you begin, ensure that you have:

• An EMR cluster running EMR V3.32.0 or later

Set up the environment

In EMR V3.32.0 and later, Hudi dependencies are built into Spark, Hive, and Presto. No additional runtime dependencies are required. Add the following Maven dependency to your pom.xml:

<dependency>
  <groupId>org.apache.hudi</groupId>
  <artifactId>hudi-spark_2.11</artifactId>
  <version>${hudi_version}</version>
  <scope>provided</scope>
</dependency>

For Spark 3, use hudi-spark_2.12 as the artifactId instead of hudi-spark_2.11.

The Hudi version bundled with your cluster depends on the EMR version. Use the following table to find the correct ${hudi_version} value:

Write data

All write examples share a common set of Hudi options. Define them once and pass them to each write operation:

val spark = SparkSession
  .builder()
  .master("local[*]")
  .appName("hudi test")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()

import spark.implicits._

// Sample dataset
val df = (for (i <- 0 until 10) yield (i, s"a$i", 30 + i * 0.2, 100 * i + 10000, s"p${i % 5}"))
  .toDF("id", "name", "price", "version", "dt")

// Common options shared across all write operations
val hudiOptions = Map(
  TABLE_NAME                             -> "hudi_test_0",
  RECORDKEY_FIELD_OPT_KEY               -> "id",
  PRECOMBINE_FIELD_OPT_KEY              -> "version",
  KEYGENERATOR_CLASS_OPT_KEY            -> classOf[SimpleKeyGenerator].getName,
  HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY -> classOf[MultiPartKeysValueExtractor].getCanonicalName,
  PARTITIONPATH_FIELD_OPT_KEY           -> "dt",
  HIVE_PARTITION_FIELDS_OPT_KEY         -> "ds",
  META_SYNC_ENABLED_OPT_KEY             -> "true",
  HIVE_USE_JDBC_OPT_KEY                 -> "false",
  HIVE_DATABASE_OPT_KEY                 -> "default",
  HIVE_TABLE_OPT_KEY                    -> "hudi_test_0"
)

Key parameters:

df.write.format("hudi")
  .options(hudiOptions)
  .option(OPERATION_OPT_KEY,  INSERT_OPERATION_OPT_VAL)
  .option(INSERT_PARALLELISM, "8")
  .option(UPSERT_PARALLELISM, "8")
  .mode(Overwrite)
  .save("/tmp/hudi/h0")

df.write.format("hudi")
  .options(hudiOptions)
  .option(OPERATION_OPT_KEY,  DELETE_OPERATION_OPT_VAL)
  .option(DELETE_PARALLELISM, "8")
  .mode(Append)
  .save("/tmp/hudi/h0")

Query data

Hudi is integrated into Spark, Hive, and Presto in EMR. No additional dependencies are needed to run queries.

To query a Hudi table with Hive or Presto, enable metadata synchronization at write time by setting META_SYNC_ENABLED_OPT_KEY to "true". See Write data.