Doris Integration with Flink Jar Sample Program

Example Code

Modify the following parameters in the com/huawei/doris/FlinkDorisConnectorJarDemo.java sample code based on the site requirements:

• HOST: IP address of the master FE node of Doris. To obtain the IP address of the master FE node, log in to FusionInsight Manager, choose Cluster > Services > Doris, and check Host Where Leader Locates.
• PORT: HTTPS port of the Doris FE service. To obtain the port, log in to FusionInsight Manager, choose Cluster > Services > Doris, click Configurations, and search for https_port.
• FE_NODES: The format is HOST1:PORT1,HOST2:PORT2,HOST3:PORT3. FE_NODES can be a single HOST:PORT or multiple sets of HOST:PORT, separated by commas.
• DATABASE: database of the table storing imported data.
• TABLE: name of the Doris table that stores imported data.
• USER: Username for the Doris database.
• PASSWD: Password for the Doris database.

/**
 * Table creation statement.
 * create database example_db;
 * CREATE TABLE `example_table` (
 * `city` varchar(256) NULL,
 * `longitude` double NULL,
 * `latitude` double NULL,
 * `destroy_date` date NULL
 * ) ENGINE=OLAP
 * DUPLICATE KEY(`city`)
 * DISTRIBUTED BY HASH(`city`) BUCKETS 3
 */
private static String DATABASE = "example_db";
private static String TABLE_NAME = "example_table";
private static String FE_NODES = ""; // Leader Node host
private static String USER = "";
private static String PASSWD = "";

public static void main(String[] args) throws Exception {
    Properties confProperties = new Properties();
    // Use ClassLoader to load the properties configuration file and generate the corresponding input stream.
    InputStream in = FlinkDorisConnectorJarDemo.class.getClassLoader().getResourceAsStream("conf.properties");
    // Load the input stream using the properties object.
    confProperties.load(in);
    //Get the value corresponding to the key
    USER = confProperties.getProperty("USER");
    PASSWD = confProperties.getProperty("PASSWD");
    FE_NODES = confProperties.getProperty("FE_NODES");

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    EnvironmentSettings environmentSettings =
            EnvironmentSettings.newInstance().inStreamingMode().build();
    env.enableCheckpointing(10);
    env.setParallelism(1);

    // enable checkpoint
    env.enableCheckpointing(10000);
    // using batch mode for bounded data
    env.setRuntimeMode(RuntimeExecutionMode.BATCH);

    //doris sink option
    DorisSink.Builder<RowData> builder = DorisSink.builder();
    DorisOptions.Builder dorisBuilder = DorisOptions.builder();

    dorisBuilder.setFenodes(FE_NODES)
            .setTableIdentifier(DATABASE + "." + TABLE_NAME)
            .setUsername(USER)
            .setPassword(PASSWD)
            // Kerberos authentication is disabled for the cluster (the cluster is in normal mode) as false, Kerberos authentication is enabled for the cluster (the cluster is in security mode) as true.
            .setIgnoreHttpsCA(true)
            // Kerberos authentication is disabled for the cluster (the cluster is in normal mode) as false, Kerberos authentication is enabled for the cluster (the cluster is in security mode) as true.
            .setEnableHttps(true)
            // The default value is true. If Flink Jar reports error 307 after running, you can change this value to false.
            .setAutoRedirect(false);

    // json format to streamload
    Properties properties = new Properties();

    properties.setProperty("format", "json");
    properties.setProperty("read_json_by_line", "false");
    DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
    executionBuilder.setLabelPrefix("flink-label-doris") //streamload label prefix
            .setDeletable(false)
            .setStreamLoadProp(properties); //streamload params

    //flink rowdata's schema
    String[] fields = {"city", "longitude", "latitude", "destroy_date"};
    DataType[] types = {DataTypes.VARCHAR(256), DataTypes.DOUBLE(), DataTypes.DOUBLE(), DataTypes.DATE()};

    builder.setDorisReadOptions(DorisReadOptions.builder().build())
            .setDorisExecutionOptions(executionBuilder.build())
            .setSerializer(RowDataSerializer.builder()    //serialize according to rowdata
                    .setFieldNames(fields)
                    .setType("json")           //json format
                    .setFieldType(types).build())
            .setDorisOptions(dorisBuilder.build());

    //mock rowdata source
    DataStream<RowData> source = env.fromElements("")
            .map(new MapFunction<String, RowData>() {
                @Override
                public RowData map(String value) throws Exception {
                    GenericRowData genericRowData = new GenericRowData(4);
                    genericRowData.setField(0, StringData.fromString("ULC"));
                    genericRowData.setField(1, 21.59);
                    genericRowData.setField(2, 31.56);
                    genericRowData.setField(3, LocalDate.now().toEpochDay());
                    return genericRowData;
                }
            });
    source.sinkTo(builder.build());
    env.execute("Flink DataStream example");