Submitting a Flink SQL Job in DLI Using Hudi
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.
Log in to the DLI management console. In the navigation pane on the left, choose Job Management > Flink Jobs.
- 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.
Feedback
Was this page helpful?
Provide feedbackThank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.
For any further questions, feel free to contact us through the chatbot.
Chatbot