使用 Insert 方式同步数据

用户可以通过 MySQL 协议，使用 INSERT 语句进行数据导入。

INSERT 语句的使用方式和 MySQL 等数据库中 INSERT 语句的使用方式类似。 INSERT 语句支持以下两种语法：

  1. * INSERT INTO table SELECT ...
  2. * INSERT INTO table VALUES(...)

这里我们仅介绍第二种方式。关于 INSERT 命令的详细说明，请参阅 INSERT 命令文档。

单次写入

单次写入是指用户直接执行一个 INSERT 命令。示例如下：

  1. INSERT INTO example_tbl (col1, col2, col3) VALUES (1000, "test", 3.25);

对于 Doris 来说，一个 INSERT 命令就是一个完整的导入事务。

因此不论是导入一条数据，还是多条数据，我们都不建议在生产环境使用这种方式进行数据导入。高频次的 INSERT 操作会导致在存储层产生大量的小文件，会严重影响系统性能。

该方式仅用于线下简单测试或低频少量的操作。

或者可以使用以下方式进行批量的插入操作：

  1. INSERT INTO example_tbl VALUES
  2. (1000, "baidu1", 3.25)
  3. (2000, "baidu2", 4.25)
  4. (3000, "baidu3", 5.25);

我们建议一批次插入条数在尽量大，比如几千甚至一万条一次。或者可以通过下面的程序的方式，使用 PreparedStatement 来进行批量插入。

JDBC 示例

这里我们给出一个简单的 JDBC 批量 INSERT 代码示例：

  1. package demo.doris;
  3. import java.sql.Connection;
  4. import java.sql.DriverManager;
  5. import java.sql.PreparedStatement;
  6. import java.sql.SQLException;
  8. public class DorisJDBCDemo {
  10.     private static final String JDBC_DRIVER = "com.mysql.jdbc.Driver";
  11.     private static final String DB_URL_PATTERN = "jdbc:mysql://%s:%d/%s?rewriteBatchedStatements=true";
  12.     private static final String HOST = "127.0.0.1"; // Leader Node host
  13.     private static final int PORT = 9030;   // query_port of Leader Node
  14.     private static final String DB = "demo";
  15.     private static final String TBL = "test_1";
  16.     private static final String USER = "admin";
  17.     private static final String PASSWD = "my_pass";
  19.     private static final int INSERT_BATCH_SIZE = 10000;
  21.     public static void main(String[] args) {
  22.         insert();
  23.     }
  25.     private static void insert() {
  26.         // 注意末尾不要加 分号 ";"
  27.         String query = "insert into " + TBL + " values(?, ?)";
  28.         // 设置 Label 以做到幂等。
  29.         // String query = "insert into " + TBL + " WITH LABEL my_label values(?, ?)";
  31.         Connection conn = null;
  32.         PreparedStatement stmt = null;
  33.         String dbUrl = String.format(DB_URL_PATTERN, HOST, PORT, DB);
  34.         try {
  35.             Class.forName(JDBC_DRIVER);
  36.             conn = DriverManager.getConnection(dbUrl, USER, PASSWD);
  37.             stmt = conn.prepareStatement(query);
  39.             for (int i =0; i < INSERT_BATCH_SIZE; i++) {
  40.                 stmt.setInt(1, i);
  41.                 stmt.setInt(2, i * 100);
  42.                 stmt.addBatch();
  43.             }
  45.             int[] res = stmt.executeBatch();
  46.             System.out.println(res);
  47.         } catch (Exception e) {
  48.             e.printStackTrace();
  49.         } finally {
  50.             try {
  51.                 if (stmt != null) {
  52.                     stmt.close();
  53.                 }
  54.             } catch (SQLException se2) {
  55.                 se2.printStackTrace();
  56.             }
  57.             try {
  58.                 if (conn != null) conn.close();
  59.             } catch (SQLException se) {
  60.                 se.printStackTrace();
  61.             }
  62.         }
  63.     }
  64. }

请注意以下几点：

  1. JDBC 连接串需添加 rewriteBatchedStatements=true 参数，并使用 PreparedStatement 方式。

    目前 Doris 暂不支持服务器端的 PrepareStatemnt，所以 JDBC Driver 会在客户端进行批量 Prepare。

    rewriteBatchedStatements=true 会确保 Driver 执行批处理。并最终形成如下形式的 INSERT 语句发往 Doris：

    1. INSERT INTO example_tbl VALUES
    2. (1000, "baidu1", 3.25)
    3. (2000, "baidu2", 4.25)
    4. (3000, "baidu3", 5.25);

  2. 批次大小

    因为是在客户端进行批量处理，所以一批次过大的话，会占用客户端的内存资源，需关注。

    Doris 后续会支持服务端的 PrepareStatemnt，敬请期待。

  3. 导入原子性

    和其他到导入方式一样，INSERT 操作本身也支持原子性。每一个 INSERT 操作都是一个导入事务，能够保证一个 INSERT 中的所有数据原子性的写入。

    前面提到，我们建议在使用 INSERT 导入数据时，采用 ”批“ 的方式进行导入，而不是单条插入。

    同时，我们可以为每次 INSERT 操作设置一个 Label。通过 Label 机制 可以保证操作的幂等性和原子性，最终做到数据的不丢不重。关于 INSERT 中 Label 的具体用法，可以参阅 INSERT 文档。