Flink-doris-connector sql写入doris on zeppelin优化版

拼搏现实的明天。 2022-10-16 07:57 529阅读 0赞

# **要是觉得写得不错，请务必点个赞，谢谢啊！！！** #

**目录**

要是觉得写得不错，请务必点个赞，谢谢啊！！！

一，代码开发

1，写入doris的案例代码

2，指定DorisDynamicTableSourceFactory

3，DorisDynamicTableSourceFactory 具体实现

4，DorisSinkFunction具体实现

5，队列类实现

6，doris sink代码，这个比较简单。注意传参就ok了。

二，打包细节

三，zeppelin执行

1，引入依赖：

2，创建语句

3，多个sql执行

四，执行效果

五，总结

--------------------

# 一，代码开发 #

## 1，写入doris的案例代码 ##

public class DorisSqlTest2 {
        private  static final String DORIS_SQL =  "CREATE TABLE doris_test (\n"+
                " order_number STRING,\n" +
                " order_key STRING,\n" +
                " stock_order STRING,\n" +
                " stock_code STRING,\n" +
                " intermediary STRING,\n" +
                " intermediary_name STRING,\n" +
                " intermediary_accountid STRING,\n" +
                " intermediary_phone STRING,\n" +
                " canal_type STRING\n" +
                " ) WITH (\n"+
                "   'connector' = 'doris',\n"+
                "   'doris.host' = '192.168.6.143',\n"+
                "   'doris.port' = '8030',\n"+
                "   'database-name' = 'example_db',\n"+
                "   'table-name' = 'assure_orders2',\n"+
                "   'username' = 'root',\n"+
                "   'password' = 'root',\n"+
                "   'max.batch' = '500'\n"+
                " )";
    
    
    
        private static final String DATA_GEN =  "CREATE TABLE datagen (\n" +
                " id STRING,\n" +
                " name STRING,\n" +
                " user_age INT,\n" +
                " user_other STRING,\n" +
                " ts AS localtimestamp\n" +
                ") WITH (\n" +
                " 'connector' = 'datagen',\n" +
                " 'rows-per-second'='500',\n" +
                " 'fields.id.length'='7',\n" +
                " 'fields.user_age.min'='1',\n" +
                " 'fields.user_age.max'='100',\n" +
                " 'fields.name.length'='2',\n" +
                " 'fields.user_other.length'='10'\n" +
                ")";
    
    
    
        private  static final String KAFKA_SQL = "CREATE TABLE kafka_test (\n" +
               " order_number STRING,\n" +
                " order_key STRING,\n" +
                " stock_order STRING,\n" +
                " stock_code STRING,\n" +
                " intermediary STRING,\n" +
                " intermediary_name STRING,\n" +
                " intermediary_accountid STRING,\n" +
                " intermediary_phone STRING,\n" +
                " canal_type STRING\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'dwd.rds_core.plateform_stable.assure_orders',\n" +
                " 'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',\n" +
                " 'properties.group.id' = 'testGroup',\n" +
                " 'format' = 'json',\n" +
                " 'scan.startup.mode' = 'earliest-offset'\n" +
                ")";
    
        public static void main(String[] args) {
    
    //        System.out.println("DORIS_SQL = " + DORIS_SQL);
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(3);
            env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
    
            EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
    //		StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, bsSettings);
            StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env);
            bsTableEnv.executeSql(DORIS_SQL);
            bsTableEnv.executeSql(DATA_GEN);
            bsTableEnv.executeSql(KAFKA_SQL);
    
            bsTableEnv.executeSql("insert into doris_test select order_number,order_key,stock_order,stock_code,intermediary,intermediary_name,intermediary_accountid,intermediary_phone,canal_type from kafka_test");
    //        bsTableEnv.executeSql("insert into kafkaTable select id,name from datagen");
        }
    }

## 2，指定DorisDynamicTableSourceFactory ##

创建相对应的路径，并且创建类

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

## 3，DorisDynamicTableSourceFactory 具体实现 ##

对应的就是sql里面的参数和类型

package org.apache.flink.connector.doris.table;
    
    import org.apache.flink.configuration.ConfigOption;
    import org.apache.flink.configuration.ConfigOptions;
    import org.apache.flink.configuration.ReadableConfig;
    import org.apache.flink.table.api.TableSchema;
    import org.apache.flink.table.connector.sink.DynamicTableSink;
    import org.apache.flink.table.factories.DynamicTableSinkFactory;
    import org.apache.flink.table.factories.FactoryUtil;
    import org.apache.flink.table.utils.TableSchemaUtils;
    
    import java.time.Duration;
    import java.util.HashSet;
    import java.util.Set;
    
    /**
     * @program: flink-neiwang-dev
     * @description: 1
     * @author: Mr.Wang
     * @create: 2020-11-12 14:04
     **/
    public class DorisDynamicTableSourceFactory implements DynamicTableSinkFactory {
        //todo 名称叫doris
        public static final String IDENTIFIER = "doris";
    
        public static final ConfigOption<String> DORIS_HOST = ConfigOptions
                .key("doris.host")
                .stringType()
                .noDefaultValue()
                .withDescription("the doris database url.");
    
    
        public static final ConfigOption<String> DORIS_HTTP_PORT = ConfigOptions
                .key("doris.port")
                .stringType()
                .noDefaultValue()
                .withDescription("the doris database url.");
    
    
        public static final ConfigOption<String> DATABASE_NAME = ConfigOptions
                .key("database-name")
                .stringType()
                .noDefaultValue()
                .withDescription("the database name.");
    
    
        public static final ConfigOption<String> TABLE_NAME = ConfigOptions
                .key("table-name")
                .stringType()
                .noDefaultValue()
                .withDescription("the jdbc table name.");
        public static final ConfigOption<String> USERNAME = ConfigOptions
                .key("username")
                .stringType()
                .noDefaultValue()
                .withDescription("the jdbc user name.");
        public static final ConfigOption<String> PASSWORD = ConfigOptions
                .key("password")
                .stringType()
                .noDefaultValue()
                .withDescription("the jdbc password.");
    
        public static final ConfigOption<Long> WINDOW_TIME = ConfigOptions
                .key("window")
                .longType()
                .defaultValue(10000L)
                .withDescription("window time.");
    
        public static final ConfigOption<Integer> MAX_BATCH_NUMBER = ConfigOptions
                .key("max.batch")
                .intType()
                .defaultValue(500)
                .withDescription("batch max number.");
    
    
        public static final ConfigOption<String> LOCAL = ConfigOptions
                .key("local")
                .stringType()
                .noDefaultValue()
                .withDescription("run in local or not.");
    
    
        @Override
        public DynamicTableSink createDynamicTableSink(Context context) {
            final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
            helper.validate();
            ReadableConfig readableConfig = helper.getOptions();
            TableSchema physicalSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
    
            //todo 调用 DorisDynamicTableSink ,传入参数
            return new DorisDynamicTableSink(readableConfig, physicalSchema);
    
        }
    
        @Override
        public String factoryIdentifier() {
            //todo 设置名称
            return IDENTIFIER;
        }
    
        @Override
        public Set<ConfigOption<?>> requiredOptions() {
    
    
            //todo 必要的参数
            Set<ConfigOption<?>> requiredOptions = new HashSet();
            requiredOptions.add(DORIS_HOST);
            requiredOptions.add(DORIS_HTTP_PORT);
            requiredOptions.add(DATABASE_NAME);
            requiredOptions.add(TABLE_NAME);
            requiredOptions.add(USERNAME);
            requiredOptions.add(PASSWORD);
            requiredOptions.add(MAX_BATCH_NUMBER);
            return requiredOptions;
    
        }
    
        @Override
        public Set<ConfigOption<?>> optionalOptions() {
            //todo 可选参数
            Set<ConfigOption<?>> optionalOptions = new HashSet<>();
            optionalOptions.add(LOCAL);
            return optionalOptions;
        }
    }

下面接着就是对应的数据sink函数

## 4，DorisSinkFunction具体实现 ##

import com.alibaba.fastjson.JSONObject;
    import com.sjb.flink.dwd.test.dorisDev_2021_03_01.DorisStreamLoadLocalTest2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.configuration.ReadableConfig;
    import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
    import org.apache.flink.table.data.GenericRowData;
    import org.apache.flink.table.data.RowData;
    import org.apache.flink.table.types.DataType;
    
    
    public class DorisSinkFunction extends RichSinkFunction<RowData> {
        private String[] fieldNames;
        private DataType[] fieldDataTypes;
        private ReadableConfig options;
        private int aa = 0;
        private DorisStreamLoadLocalTest2 dorisStreamLoad;
        private DataTransferRunner dataTransferRunner;
        private transient JSONObject json;
    
    
        public static final int TIMEOUT = 60000;
    
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
    
    
            this.dorisStreamLoad = new DorisStreamLoadLocalTest2(options);
    
            Integer MAX_BATCH_NUMBER = options.get(DorisDynamicTableSourceFactory.MAX_BATCH_NUMBER);
            this.dataTransferRunner = new DataTransferRunner(MAX_BATCH_NUMBER, dorisStreamLoad);
            Thread thread = new Thread(dataTransferRunner);
            thread.start();
        }
    
        public DorisSinkFunction(ReadableConfig readableConfig, DataType[] dataType, String[] fieldNamesArray) {
            this.options = readableConfig;
            this.fieldDataTypes = dataType;
            this.fieldNames = fieldNamesArray;
    
        }
    
    
        @Override
        public void invoke(RowData value, Context context) throws Exception {
    
            try {
                json = new JSONObject();
                for (int i = 0; i < fieldNames.length; i++) {
                    String dataType = fieldDataTypes[i].toString();
                    String fieldName = fieldNames[i];
    
                    if (dataType.toLowerCase().equals("int")) {
                        json.put(fieldName, value.getInt(i));
                    } else if (dataType.toLowerCase().equals("long")) {
                        json.put(fieldName, value.getLong(i));
    
                    } else if (dataType.toLowerCase().equals("double")) {
                        json.put(fieldName, value.getDouble(i));
                    } else if (dataType.toLowerCase().equals("float")) {
                        json.put(fieldName, value.getFloat(i));
                    } else {
                        GenericRowData rowData = (GenericRowData) value;
                        if (null != rowData.getField(i)) {
                            json.put(fieldName, rowData.getField(i).toString());
    
                        }
                    }
    
                }
    
                aa++;
                if (aa % 1000 ==0){
                    System.out.println("计数  = " + aa);
                }
    
    
                dataTransferRunner.sendData(json);
    
                if (dataTransferRunner.hasException()) {
                    //todo 任务异常
                    System.out.println("%%%%%%%%%%%%%%%%%%%%%%%任务异常。。。");
                    Exception exception = dataTransferRunner.getException();
                    exception.printStackTrace();
    
                }
    
            } catch (Exception ex) {
                ex.printStackTrace();
                throw new Exception(ex.getMessage() + ",异常,任务失败，时间: " + System.currentTimeMillis());
         /*       String clazzName = this.getClass().getName();
                String class_name = clazzName.substring(clazzName.lastIndexOf('.') + 1,clazzName.lastIndexOf('$'));
    
                String content = "[" + class_name + "],测试异常:{" + ex.getMessage() + "},任务失败,时间: " + System.currentTimeMillis();
    //                DingProd.sendDingRabotTest(content);
                throw new Exception(ex.getMessage() + ",异常,任务失败，时间: " + System.currentTimeMillis());*/
            }
    
        }
    
    
    }

注意：

1，能在open()方法里面创建对象，对象初始化的就在open方法里面做。

2，这是跟flink 的底层序列化有关系，遇到的坑就是本地执行没问题，在集群执行有问题，报序列化的问题。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

3，这里是使用的阻塞队列

## 5，队列类实现 ##

import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONArray;
    import com.alibaba.fastjson.JSONObject;
    import com.sjb.flink.dwd.test.dorisDev_2021_03_01.DorisStreamLoadLocalTest2;
     
    
    import java.io.Serializable;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    public class DataTransferRunner implements Runnable, Serializable {
    
        private static final long serialVersionUID = 1L;
        private volatile boolean continued = true;
        private Exception exception = null;
        private Integer BATCH_SIZE = 500;
        private LinkedBlockingQueue<JSONObject> linkedBlockingQueue = new LinkedBlockingQueue<JSONObject>(BATCH_SIZE * 3);
        private List<JSONObject> tmpList = new ArrayList<JSONObject>(BATCH_SIZE);
    
        private DorisStreamLoadLocalTest2 dorisStreamLoad;
        private  JSONArray jsonArray;
    
        public DataTransferRunner(Integer maxBatch, DorisStreamLoadLocalTest2 dorisStreamLoad) {
            this.dorisStreamLoad = dorisStreamLoad;
            this.BATCH_SIZE = maxBatch;
    
        }
    
        @Override
        public void run() {
    
            while (this.continued) {
                try {
                    tmpList.clear();
                    this.linkedBlockingQueue.drainTo(tmpList, BATCH_SIZE);
                    while (this.continued && tmpList.isEmpty()) {
                        //todo 队列没数据 睡眠
                        Thread.sleep(200);
                        this.linkedBlockingQueue.drainTo(tmpList, BATCH_SIZE);
                    }
    
                    jsonArray = JSONArray.parseArray(JSON.toJSONString(tmpList));
                    dorisStreamLoad.batchInsertDoris(jsonArray.toJSONString());
    
                } catch (final Exception exception) {
                    System.out.println("进入异常.............");
                    this.exception = exception;
                    break;
                }
            }
        }
    
    
        public void sendData(JSONObject value) throws InterruptedException {
    
            int offerTryCount = 0;
            //todo 后端消费不过来 队列写不进去，重试2秒之后 还不能写入，会返回true ，然后进行下面的操作 阻塞{}秒。
            while (!linkedBlockingQueue.offer(value, 1000, TimeUnit.MILLISECONDS)) {
    
                if (++offerTryCount % 10 == 0) {
                    //todo 下游消费不过来,代表等待了10秒了，要进行下一步的处理
                    this.exception = new RuntimeException("下游消费不过来,代表等待了10秒了，要进行下一步的处理\"");
    
                }
            }
        }
    
        public boolean hasException() {
            return this.exception != null;
        }
    
        public Exception getException() {
            return this.exception;
        }
    
    }

## 6，doris sink代码，这个比较简单。注意传参就ok了。 ##

import com.sjb.common.exceptions.BizException;
    import org.apache.commons.codec.binary.Base64;
    import org.apache.commons.lang3.StringUtils;
    import org.apache.flink.configuration.ReadableConfig;
    import org.apache.flink.connector.doris.table.DorisDynamicTableSourceFactory;
    import org.apache.http.HttpEntity;
    import org.apache.http.HttpHeaders;
    import org.apache.http.HttpVersion;
    import org.apache.http.client.config.RequestConfig;
    import org.apache.http.client.methods.CloseableHttpResponse;
    import org.apache.http.client.methods.HttpPut;
    import org.apache.http.entity.StringEntity;
    import org.apache.http.impl.client.CloseableHttpClient;
    import org.apache.http.impl.client.DefaultRedirectStrategy;
    import org.apache.http.impl.client.HttpClientBuilder;
    import org.apache.http.impl.client.HttpClients;
    import org.apache.http.util.EntityUtils;
    
    import java.io.Serializable;
    import java.nio.charset.StandardCharsets;
    
    
    public class DorisStreamLoadLocalTest2 implements Serializable {
        private static final long serialVersionUID = 1L;
    
    
        private static final String CHARSET = "UTF-8";
    
        private static final Integer HTTP_OK = 200;
    
        private static final String CONTENT_TYPE_NAME = "Content-Type";
    
        private static final String CONTENT_TYPE_JSON = "application/json;charset=UTF-8";
    
        private static final String CONTENT_TYPE_XML = "text/xml;charset=UTF-8";
    
        private static final String CONTENT_TYPE_FORM = "application/x-www-form-urlencoded;charset=UTF-8";
    
        private static final String ACCEPT_NAME = "Accept";
        private static final String ACCEPT = "application/json;charset=UTF-8";
        private static ReadableConfig options;
    
        //这个要弄长点
        public static final int TIMEOUT = 60000;
        private static String loadUrl;
    
        private static HttpClientBuilder httpClientBuilder = HttpClients
                .custom()
                .setRedirectStrategy(new DefaultRedirectStrategy() {
                    @Override
                    protected boolean isRedirectable(String method) {
                        return true;
                    }
                });
    
        private static RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(TIMEOUT)
                .setConnectTimeout(TIMEOUT)
                .setConnectionRequestTimeout(TIMEOUT)
                .build();
    
    
        public DorisStreamLoadLocalTest2() {
    
        }
    
        public DorisStreamLoadLocalTest2(ReadableConfig ReadableConfig) {
            options = ReadableConfig;
    
            String DORIS_HOST = options.get(DorisDynamicTableSourceFactory.DORIS_HOST);
            String DORIS_HTTP_PORT = options.get(DorisDynamicTableSourceFactory.DORIS_HTTP_PORT);
            String DATABASE_NAME = options.get(DorisDynamicTableSourceFactory.DATABASE_NAME);
            String TABLE_NAME = options.get(DorisDynamicTableSourceFactory.TABLE_NAME);
    
            loadUrl = String.format("http://%s:%s/api/%s/%s/_stream_load",
                    DORIS_HOST,
                    DORIS_HTTP_PORT,
                    DATABASE_NAME,
                    TABLE_NAME);
        }
    
    
        public Boolean batchInsertDoris(String content) {
    
            CloseableHttpClient client = httpClientBuilder.build();
    
    //        String DORIS_HOST = "dev-hadoop-ct7-6-143";
    //        String DORIS_HTTP_PORT = "8030";
    //        String DATABASE_NAME = "example_db";
    //        String TABLE_NAME = "assure_orders2";
    //        String USERNAME ="root";
    //        String PASSWORD = "root";
    
            String loadResult = "";
            CloseableHttpResponse response = null;
    
    
            try {
    
    
                HttpPut httpPut = new HttpPut(loadUrl);
                StringEntity entity = new StringEntity(content, "UTF-8");
                httpPut.setHeader(HttpHeaders.EXPECT, "100-continue");
                String USERNAME = options.get(DorisDynamicTableSourceFactory.USERNAME);
                String PASSWORD = options.get(DorisDynamicTableSourceFactory.PASSWORD);
                httpPut.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(USERNAME, PASSWORD));
                httpPut.setHeader("strip_outer_array", "true");
                httpPut.setHeader("format", "json");
                httpPut.setHeader("merge_type", "MERGE");
                httpPut.setHeader("delete", "canal_type=\"DELETE\"");
                httpPut.setProtocolVersion(HttpVersion.HTTP_1_0);
                httpPut.addHeader(CONTENT_TYPE_NAME, CONTENT_TYPE_FORM);
                httpPut.addHeader(ACCEPT_NAME, ACCEPT);
    //            httpPut.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_CLOSE);
    
    
                //todo 输出参数
                httpPut.setConfig(requestConfig);
                httpPut.setEntity(entity);
                response = client.execute(httpPut);
                HttpEntity httpEntity = response.getEntity();
    
                //todo 调用方法
                if (httpEntity != null) {
                    loadResult = EntityUtils.toString(httpEntity);
                }
                final int statusCode = response.getStatusLine().getStatusCode();
    
                EntityUtils.consume(httpEntity);
    
                if (statusCode != 200) {
    
                    System.out.println("loadResult = " + loadResult);
                    throw new BizException("写入失败，loadResult=" + loadResult);
                } else {
                    if (loadResult.contains("OK") && loadResult.contains("Success")) {
                        //todo 正式环境要去掉
                        System.out.println("写入成功，loadResult = " + loadResult);
                        return true;
    //                    throw new BizException("[DorisStreamLoadProd]出错，异常信息：[ 1111],具体错误地址: [" + 1111 + "]");
                    } else {
                        System.out.println("写入失败，loadResult=" + loadResult);
                        throw new BizException("写入失败，loadResult=" + loadResult);
    
                    }
                }
    
    
            } catch (Exception ex) {
                // 数组转字符串(逗号分隔)(推荐)
                String error_info = StringUtils.join(ex.getStackTrace(), "\n");
                throw new BizException("[DorisStreamLoadLocalTest2]出错，异常信息：[" + ex.toString() + "],具体错误地址: [" + error_info + "]");
    
            } finally {
    
                try {
                    if (null != response) {
                        response.close();
                    }
                    if (null != client) {
                        client.close();
                    }
    
    
                } catch (Exception ex) {
                    String error_info = StringUtils.join(ex.getStackTrace(), "\n");
                    throw new BizException("[DorisStreamLoadProd],response关闭异常,异常信息：[" + ex.toString() + "],具体错误地址: [" + error_info + "]");
                }
    
            }
    
    
        }
    
    
        private String basicAuthHeader(String username, String password) {
            final String tobeEncode = username + ":" + password;
            byte[] encoded = Base64.encodeBase64(tobeEncode.getBytes(StandardCharsets.UTF_8));
            return "Basic " + new String(encoded);
        }
      
    
    }

# 二，打包细节 #

1，在集群运行找不到

org.apache.flink.connector.doris.table.DorisDynamicTableSourceFactory

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

解决方案，在pom文件加上：

<build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.8.1</version>
                    <configuration>
                        <encoding>UTF-8</encoding>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-shade-plugin</artifactId>
                    <version>2.4.3</version>
                    <executions>
                        <execution>
                            <phase>package</phase>
                            <goals>
                                <goal>shade</goal>
                            </goals>
                            <configuration>
                                <filters>
                                    <filter>
                                        <artifact>*:*</artifact>
                                        <excludes>
                                            <exclude>META-INF/*.SF</exclude>
                                            <exclude>META-INF/*.DSA</exclude>
                                            <exclude>META-INF/*.RSA</exclude>
                                        </excludes>
                                    </filter>
                                </filters>
                                <artifactSet>
                                    <excludes>
                                        <exclude>junit:junit</exclude>
                                    </excludes>
                                </artifactSet>
                                <transformers>
                                    <transformer
                                            implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                </transformers>
    
                            </configuration>
    
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>

将jar包放入flink集群 目录。

# 三，zeppelin执行     #

## 1，引入依赖： ##

%flink.conf
    flink.execution.packages org.apache.flink:flink-connector-kafka_2.11:1.11.2,com.alibaba:fastjson:1.2.60
    
    flink.execution.jars /wyyt/software/1.11.2-flink/flink-1.0-SNAPSHOT.jar

## 2，创建语句 ##

%flink.ssql
    
    DROP TABLE IF EXISTS doris_test;
    
    --创建kafka表
    CREATE TABLE doris_test (
    order_number STRING,
    order_key STRING,
    stock_order STRING,
    stock_code STRING,
    intermediary STRING,
    intermediary_name STRING,
    intermediary_accountid STRING,
    intermediary_phone STRING,
    canal_type STRING
    ) WITH (
    'connector' = 'doris',
    'doris.host' = '192.168.6.143',
    'doris.port' = '8030',
    'database-name' = 'example_db',
    'table-name' = 'assure_orders',
    'username' = 'root',
    'password' = 'root',
    'max.batch' = '500'
    );
    
    
    
    DROP TABLE IF EXISTS doris_test2;
    
    --创建kafka表
    CREATE TABLE doris_test2 (
    order_number STRING,
    order_key STRING,
    stock_order STRING,
    stock_code STRING,
    intermediary STRING,
    intermediary_name STRING,
    intermediary_accountid STRING,
    intermediary_phone STRING,
    canal_type STRING
    ) WITH (
    'connector' = 'doris',
    'doris.host' = '192.168.6.143',
    'doris.port' = '8030',
    'database-name' = 'example_db',
    'table-name' = 'assure_orders2',
    'username' = 'root',
    'password' = 'root',
    'max.batch' = '500'
    );
    
    
    DROP TABLE IF EXISTS kafka_test;
     
    CREATE TABLE kafka_test (
    order_key    string ,    
    order_number string ,    
    stock_order  string ,    
    stock_code   string ,    
    intermediary string ,    
    intermediary_name  string ,    
    intermediary_accountid  string ,    
    intermediary_phone string ,    
    tradeid_by_intermediary_state   int  ,  
    tradeid_by_driver_state int    ,    
    driver  string ,    
    driver_name  string ,    
    driver_accountid   string ,    
    driver_phone string ,    
    pay_by_driver      bigint ,    
    driver_pay_time    string ,    
    pay_by_intermediary     bigint ,    
    intermediary_pay_time   string ,    
    create_date  string ,    
    tradeid_by_driver  string ,    
    tradeid_by_intermediary string ,    
    dynamic_state      int    ,    
    static_state int    ,    
    informationfees    bigint ,    
    empty_margin bigint ,    
    deal_close_time    string ,    
    driver_favorable   bigint ,    
    driver_poundage    bigint ,    
    intermediary_favorable  bigint ,    
    intermediary_poundage   bigint ,    
    consult_state      int,    
    complain_state     int,    
    suggest_state      int,    
    suggest_time string ,    
    deal_state   int    ,    
    order_kind   int    ,    
    freight_rates      bigint ,    
    deposit_rates      bigint ,    
    sign_in_state      int    ,    
    consult_dead_time  string ,    
    check_dead_time    string ,    
    finish_dead_time   string ,    
    nu      string ,    
    logistics_company  int    ,    
    credit_pay   int    ,    
    company_code string ,    
    original_intermediary   string ,    
    original_intermediary_phone     string    ,  
    is_assigned  int    ,    
    electronic_contract_no  string ,    
    oil_card     bigint ,    
    is_haulage_operator     int    ,    
    scheduling_fee     bigint ,    
    last_scheduling_fee     bigint ,    
    sign_rates   bigint ,    
    phone_num    string ,    
    device_code  string ,    
    original_unit      string ,    
    actual_unit  string ,    
    unit_price   bigint ,    
    channel_code string ,    
    company_extra_url  string ,    
    sign_load_time     string ,    
    sign_upload_time   string ,    
    client_check_time  string ,    
    company_check_time string ,    
    company_check_receipt_time string    ,  
    driver_income      bigint ,    
    update_time  string ,    
    is_settlement      int    ,    
    etc_rates    bigint ,    
    carrier_code string ,    
    leader_driver_code string ,    
    original_interme_pound  bigint ,    
    original_interme_favo   bigint ,    
    work_order_code    string ,    
    work_times   int    ,    
    order_check_state  int    ,    
    order_multi_check_state int    ,    
    agent_code   string ,    
    create_is_dr int    ,    
    sign_is_dr   int    ,    
    lending_channel    int    ,    
    carrier_level_number    int    ,    
    canal_type    string    ,    
    order_state_tag    int    
    ) WITH (
     'connector' = 'kafka',
     'topic' = 'dwd.rds_core.plateform_stable.assure_orders',
     'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',
     'format' = 'json' 
    );

## 3，多个sql执行 ##

%flink.ssql(runAsOne=true,type=update, parallelism=3,jobName='一个任务，多个插入测试')
    set table.dynamic-table-options.enabled=true ;
    
    insert into  doris_test select order_number,order_key,stock_order,stock_code,intermediary,intermediary_name,intermediary_accountid,intermediary_phone,canal_type from kafka_test /*+ OPTIONS( 'scan.startup.mode' = 'earliest-offset','properties.group.id' = 'testGroup1') */;
    insert into  doris_test2 select order_number,order_key,stock_order,stock_code,intermediary,intermediary_name,intermediary_accountid,intermediary_phone,canal_type from kafka_test /*+ OPTIONS('scan.startup.mode' = 'earliest-offset','properties.group.id' = 'testGroup2') */ ;

**注意点： 如果使用的是同一个kafka数据源，在创建kafka表的时候不要指定一些参数，在sql执行的时候通过 hint指定，如果这个不懂的小伙伴可以看官网。**

[https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/sql/hints.html][https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_table_sql_hints.html]

如果不这么做的话，插入语句只有前者在执行，后者任务启动，是没有消费数据的。

# 四，执行效果     #

我们可以清楚的看到，2个任务在一个job：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

# 五，总结     #

在zeppelin上面执行的时候一定要测试，校验。

思路就是模仿flink-mysql-connecter，底层用doris的stream load方式批量写入到doris，我代码里面用了到队列的原因就是：

1，如果上游数据源数据是有界的，某段时候如果没有数据过来的话，会导致最后一批数据没有写入到doris，所以这里要用队列，循环去取，如果没有新数据，要时间过期之后自己触发写入操作。，并且要考虑到积压的问题。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221014/4d4524f6c5794fb0be0091e3e9c17ab7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221014/36def7501a164a9baaf933589cefe131.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20221014/df5422435e0b4e7fbf4f80d7cb73ec79.png
[https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_table_sql_hints.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/sql/hints.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20221014/8284bc1982d143fab6dfbb9eb2a3bb14.png