Adding Third-Party Dependency JAR Packages to a FlinkServer Job

Uploading Dependency Packages

1. Log in to FusionInsight Manager and access the Flink web UI. For details, see Accessing the FlinkServer Web UI.
2. Click Dependency Management. The Dependency Management page is displayed.
3. Click Add Dependency.

   Table 1 Adding a dependency

   Parameter	Description	Example
   Customized connector or not	Whether the dependency is a custom connector. Set this parameter based on the site requirements. Yes: The file is a custom connector dependency. No: The file is a non-custom connector dependency.	Yes
   Name	Name of the dependency, which must be the same as the connection name of connector in the uploaded dependency package. Dependency packages with the same name cannot be uploaded.	kafka
   Register jar	Upload method of a JAR package: Upload File: Upload a JAR package form the local host. Specify Path: Upload a prepared dependency file from an HDFS path.	Uploading files
   Upload File	If Register jar is set to Upload File, select a JAR file from the local host.	-
   Specify Path	If Register jar is set to Specify Path, enter the HDFS path of the dependency file. (The JAR package must have been uploaded to the HDFS.)	/flink_upload_test/flink-connector-kafka-customization.jar
   Description	Description of the dependency to be uploaded.	-

4. Click OK

Example

• Custom connector dependencies
  1. Upload a custom connector dependency by referring to Uploading Dependency Packages.

     For example, the dependency name is kafka, and the name of the custom connector JAR package is flink-connector-kafka-customization.jar.

  2. Create a SQL job by referring to Creating a FlinkServer Job. Set connector in the SQL statement to the dependency name, for example, 'connector'='kafka'.

     CREATE TABLE KafkaSinkTable (`user_id` INT, `name` VARCHAR) WITH (
       'connector' = 'kafka',
       'topic' = 'test_sink6',
       'properties.bootstrap.servers' = '192.168.20.134:21005',
       'properties.group.id' = 'testGroup',
       'scan.startup.mode' = 'earliest-offset',
       'format' = 'csv'
     );
     CREATE TABLE datagen (`user_id` INT, `name` VARCHAR) WITH (
       'connector' = 'datagen',
       'rows-per-second' = '5',
       'fields.user_id.kind' = 'sequence',
       'fields.user_id.start' = '1',
       'fields.user_id.end' = '1000'
     );
     insert INTO
       KafkaSinkTable
     select
       *
     from
       datagen;