Doris对接Flink Jar样例程序

代码样例

需根据实际环境修改“com/huawei/doris/FlinkDorisConnectorJarDemo.java”样例代码中的以下参数值：

• HOST：值为Doris的Master FE节点IP地址，Master FE节点可通过在Manager界面，选择“集群 > 服务 > Doris”，查看“Leader所在的主机”获取。
• PORT：值为Doris FE服务的HTTPS端口，可登录FusionInsight Manager，选择“集群 > 服务 > Doris > 配置”，搜索“https_port”获取。
• FE_NODES： 格式为“HOST1:PORT1,HOST2:PORT2,HOST3:PORT3”，FE_NODES可以是一组“HOST:PORT”，也可以是多组“HOST:PORT”，使用逗号分隔。
• DATABASE：存放导入数据的表所在的数据库。
• TABLE：存放导入数据的Doris表名称。
• USER：Doris数据库的用户名。
• PASSWD：Doris数据库的密码。

/**
 * 建表语句
 * 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();
    // 使用ClassLoader加载properties配置文件生成对应的输入流
    InputStream in = FlinkDorisConnectorJarDemo.class.getClassLoader().getResourceAsStream("conf.properties");
    // 使用properties对象加载输入流
    confProperties.load(in);
    //获取key对应的value值
    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认证（普通模式）为“false”，集群已启用Kerberos认证（安全模式）为“true”
            .setIgnoreHttpsCA(true)
            // 集群未启用Kerberos认证（普通模式）为“false”，集群已启用Kerberos认证（安全模式）为“true”
            .setEnableHttps(true)
            // 默认值为“true”，如果Flink Jar在运行后报错“307”，可将该值修改为“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");