Submitting a Spark Jar Job in DLI Using Hudi

Updated on 2025-02-22 GMT+08:00

View PDF

To submit a Spark Jar job, you need to manually configure the Hudi lock provider when using LakeFormation as the metadata service. Refer to Hudi Lock Configuration.

Log in to the DLI management console and choose Job Management > Spark Jobs.

To submit a Spark Jar job related to Hudi, select Spark 3.3.1 and make sure that the general-purpose queue supports Hudi.
Click Create Job in the upper right corner.

Write and package the Spark JAR file (using a Maven project as an example).

Create or use an existing Maven Java project, and introduce dependencies for Scala 2.12, Spark 3.3.1, and Hudi 0.11.0 in the pom.xml file. Since the DLI environment already provides the required dependencies, you can set the scope to provided.

<dependencies>
  <dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>2.12.15</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>3.3.1</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.3.1</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.hudi</groupId>
    <artifactId>hudi-spark3-bundle_2.12</artifactId>
    <version>0.11.0</version>
    <scope>provided</scope>
  </dependency>
  <!-- ... -->
</dependencies>

Configure scala-maven-plugin for compilation and packaging.

<build>
  <plugins>
    <plugin>
      <groupId>net.alchim31.maven</groupId>
      <artifactId>scala-maven-plugin</artifactId>
      <version>3.3.1</version>
      <executions>
        <execution>
          <goals>
            <goal>compile</goal>
            <goal>testCompile</goal>
          </goals>
        </execution>
      </executions>
    </plugin>
    <!-- ... -->
  </plugins>
  <!-- ... -->
</build>

Then, create a Scala directory under the main directory, and create a package within it. Inside the package directory, create a Scala file and write the following:

import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.{Row, SaveMode, SparkSession}
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}

import java.util.{ArrayList, List => JList}

object HudiScalaDemo {
  def main(args: Array[String]): Unit = {
    // Step 1: Obtain or create a SparkSession instance.
    val spark = SparkSession.builder
      .enableHiveSupport
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .appName("spark_jar_hudi_demo")
      .getOrCreate

    // Step 2: Construct DataFrame data for writing.
    val schema = StructType(Array(
      StructField("id", DataTypes.IntegerType),
      StructField("name", DataTypes.StringType),
      StructField("update_time", DataTypes.StringType),
      StructField("create_time", DataTypes.StringType)
    ))
    val data: JList[Row] = new ArrayList[Row]()
    data.add(new GenericRowWithSchema(Array(1, "Alice", "2024-08-05 09:00:00", "2024-08-01"), schema))
    data.add(new GenericRowWithSchema(Array(2, "Bob", "2024-08-05 09:00:00", "2024-08-02"), schema))
    data.add(new GenericRowWithSchema(Array(3, "Charlie", "2024-08-05 09:00:00", "2024-08-03"), schema))
    val df = spark.createDataFrame(data, schema)

    // Step 3: Configure the table name and OBS path.
    val databaseName = "default"
    val tableName = "hudi_table"
    val basePath = "obs://bucket/path/hudi_table"

    // Step 4: Write data and synchronize the metadata service provided by DLI to create a table.
    df.write.format("hudi")
      .option("hoodie.table.name", tableName)
      .option("hoodie.datasource.write.table.type", "COPY_ON_WRITE")
      .option("hoodie.datasource.write.recordkey.field", schema.fields(0).name)  // Primary key, which is mandatory.
      .option("hoodie.datasource.write.precombine.field", schema.fields(2).name)  // Pre-aggregation key, which is mandatory. If not needed, configure the same column as the primary key.
      .option("hoodie.datasource.write.partitionpath.field", schema.fields(3).name)  // Partition column. Multiple partitions can be configured and separated by commas (,).
      .option("hoodie.datasource.write.keygenerator.class", "org.apache.hudi.keygen.ComplexKeyGenerator")
      // When using DLI to provide metadata service, you need to configure the corresponding Hudi lock provider.
      .option("hoodie.write.lock.provider", "com.huawei.luxor.hudi.util.DliCatalogBasedLockProvider")
      // Enable synchronization.
      .option("hoodie.datasource.hive_sync.enable", "true")
      .option("hoodie.datasource.hive_sync.partition_fields", schema.fields(3).name)
// Set this parameter based on the actual partition field. For a non-partitioned table, select org.apache.hudi.hive.NonPartitionedExtractor.
      .option("hoodie.datasource.hive_sync.partition_extractor_class", "org.apache.hudi.hive.MultiPartKeysValueExtractor")
      .option("hoodie.datasource.hive_sync.use_jdbc", "false")
      .option("hoodie.datasource.hive_sync.table", tableName)
      .option("hoodie.datasource.hive_sync.database", databaseName)
      // Select a save mode as needed.
      .mode(SaveMode.Overwrite)
      .save(basePath)

    // Step 5: Run the following SQL statement to query the table:
    spark.sql(s"select id,name,update_time,create_time from ${databaseName}.${tableName} where create_time='2024-08-01'")
      .show(100)
  }
}

Run the Maven packaging command to obtain the JAR file from the target directory and upload it to the OBS directory.

mvn clean install

Submit the Spark Jar job.

Log in to the DLI management console. In the navigation pane on the left, choose Job Management > Spark Jobs. On the displayed page, click Create Job in the upper right corner.
- Select a queue for Queues and set Spark Version to 3.3.1 or later.
- You can configure the job name to facilitate identification and filtering.
- Set Application. The path points to the Spark JAR file uploaded to OBS in the previous step.
- Configure an agency. Select the agency required for submitting DLI jobs.
- Set Main Class(--class) (Optional) to the full name of the class that contains the main function to be executed.
- You can also configure Hudi parameters in Spark Arguments(--conf), but you need to add the prefix spark.hadoop.. Here is an example:
```
spark.hadoop.hoodie.write.lock.provider=com.huawei.luxor.hudi.util.DliCatalogBasedLockProvider
```
- Set Access Metadata to Yes. You are advised to use the metadata service to manage Hudi tables. The configuration in the previous step contains the configuration item for synchronizing metadata.
Click Execute in the upper right corner to submit the job.
Execute the job and check the logs. (Note: Log archiving may take some time. Logs are typically archived within 1 to 5 minutes after the job execution.)

After you click Execute, the Spark Jobs page is displayed, where you can view the job execution status. Click More in the Operation column of the job and select an operation.
- View Log: Redirects to the OBS page where you can see the complete log archive addresses of the job, including commit logs, driver logs, and executor logs. You can download the logs here.
- Commit Logs: Redirects to the aggregated commit log display page where you can view log information during job submission.
- Driver Logs: Redirects to the aggregated display page for driver logs, sequentially displaying spark.log, stderr.log, and stdout.log from top to bottom.
Then select Driver Logs. If the logs are not yet aggregated, wait a few minutes and check again. You can see the result of the select statement printed by the sample program in stdout.log at the bottom of the logs.