CREATE TABLE IF NOT EXISTS demo.user
(
 `id`   INT NOT NULL,
 `name` VARCHAR(255),
 `age`  INT
) DISTRIBUTED BY HASH(`id`)
PROPERTIES (
 "replication_num" = "1"
);

<properties>
        <flink.version>1.12.1</flink.version>
        <scala.version>2.12.13</scala.version>
        <mysql.version>8.0.25</mysqlc.version>
        <lombok.version>1.18.12</lombok.version>
    </properties>

    <dependencies>
        <!-- 写入数据到doris -->
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
             <version>${mysql.version</version>
         </dependency>
        <!-- flink核心API -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>${lombok.version}</version>
        </dependency>
    </dependencies>

package com.daniel.bean;

import lombok.Builder;
import lombok.Data;

/**
 * @Author Daniel
 * @Date: 2023/7/3 15:35
 * @Description
 **/

@Data
@Builder
public class User {
    public int id;
    public String name;
    public int age;
}

package com.daniel.util;

import com.daniel.bean.User;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
* @Author Daniel
* @Date: 2023/7/3 15:36
* @Description
**/


public class DorisSinkFunction extends RichSinkFunction<User> {
 Connection conn = null;
 String sql;

 public DorisSinkFunction(String sql) {
     this.sql = sql;
 }

 @Override
 public void open(Configuration parameters) throws Exception {
     super.open(parameters);
     conn = getConn("localhost", 9030, "demo");
 }

 @Override
 public void close() throws Exception {
     super.close();
     if (conn != null) {
         conn.close();
     }
 }

 // 定义具体的操作
 @Override
 public void invoke(User user, Context context) throws Exception {
     // 批量插入
     PreparedStatement preparedStatement = conn.prepareStatement(sql);
     preparedStatement.setLong(1, user.id);
     preparedStatement.setString(2, user.name);
     preparedStatement.setLong(3, user.age);
     preparedStatement.addBatch();

     long startTime = System.currentTimeMillis();
     int[] batchResult = preparedStatement.executeBatch();
     long endTime = System.currentTimeMillis();
     System.out.println("批量插入用时：" + (endTime - startTime) + "ms -- 插入数据行数：" + batchResult.length);
 }

 public Connection getConn(String host, int port, String database) throws SQLException, ClassNotFoundException {
     Class.forName("com.mysql.cj.jdbc.Driver");
     String address = "jdbc:mysql://" + host + ":" + port + "/" + database;
     conn = DriverManager.getConnection(address, "root", "");
     return conn;
 }
}

package com.daniel;

import com.daniel.bean.User;
import com.daniel.util.DorisSinkFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
* @Author Daniel
* @Date: 2023/7/3 15:37
* @Description
**/

public class DorisWriteTest {
 public static void main(String[] args) throws Exception {
     StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
     // Source
     DataStream<String> ds = env.socketTextStream("localhost", 9999);

     // Transform
     SingleOutputStreamOperator<User> dataStream = ds.map((MapFunction<String, User>) data -> {
         String[] split = data.split(",");
         return User.builder()
                 .id(Integer.parseInt(split[0]))
                 .name(split[1])
                 .age(Integer.parseInt(split[2]))
                 .build();
     });

     // Sink
     String sql = "INSERT INTO demo.user (id, name, age) VALUES (?,?,?)";
     DorisSinkFunction jdbcSink = new DorisSinkFunction(sql);
     dataStream.addSink(jdbcSink);
     env.execute("flink-doris-write");
 }
}

nc -L -p 9999

1,Daniel,25
2,David,38
3,James,16
4,Robert,27

select *
 from demo.user;