Submitting a Flink SQL Job in DLI Using Hudi

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

View PDF

This section describes how to submit a Flink SQL job in DLI using Hudi.

For details about the syntax, see Flink OpenSource SQL 1.15 Syntax Overview.

Create a Flink job. Specifically, click Create Job in the upper right corner. In the dialog box that appears, set Type to Flink OpenSource SQL and enter a name.

Write Flink SQL statements (without using catalogs).

Here, the sink table points to the Hudi table path by creating a temporary table to write data, and the table parameters are configured with hive_sync related parameters to synchronize metadata to the metadata service provided by DLI in real-time. (See the Flink parameters section for specific parameters.)

Modify the path parameter of the sink table in the job to the desired OBS path for saving the Hudi table.

-- Temporary table as the source, using datagen to mock data
create table
  orderSource (
    order_id STRING,
    order_name STRING,
    order_time TIMESTAMP(3)
  )
with
  ('connector' = 'datagen', 'rows-per-second' = '1');

-- Create a Hudi temporary table as the sink, configuring hms to synchronize the table to DLI metadata service.
CREATE TABLE
  hudi_table (
    order_id STRING PRIMARY KEY NOT ENFORCED,
    order_name STRING,
    order_time TIMESTAMP(3),
    order_date String
  ) PARTITIONED BY (order_date)
WITH
  (
    'connector' = 'hudi',
    'path' = 'obs://bucket/path/hudi_table',
    'table.type' = 'MERGE_ON_READ',
    'hoodie.datasource.write.recordkey.field' = 'order_id',
    'write.precombine.field' = 'order_time',
    'hive_sync.enable' = 'true',
    'hive_sync.mode' = 'hms',
    'hive_sync.table' = 'hudi_table',
    'hive_sync.db' = 'default'
  );

-- Execute the insert statement to read from the source table and write to the sink.
insert into
  hudi_table
select
  order_id,
  order_name,
  order_time,
  DATE_FORMAT (order_time, 'yyyyMMdd')
from
  orderSource;

Set job running parameters.
- Select a queue and configure the Flink version to at least 1.15.
- Configure an agency with sufficient permissions.
- Configure an OBS bucket.
- Enable checkpointing. Checkpointing must be enabled when using Hudi.
Submit the job and check the FlinkUI and logs.

Click Submit in the upper right corner of the page. Confirm the parameters are correct on the displayed page, then click Start at the bottom. After submission, the Flink job page is displayed, where you can filter the recently submitted Flink job and check its execution status.

Click the job name to navigate to the job page. Here you can click the Commit Logs or Run Log tab to check the aggregated logs. You can also directly click the Logs tab, select JobManager or TaskManager, and download the corresponding logs.

Click More in the upper right corner of the page and select FlinkUI to navigate to the FlinkUI for the job.