E-MapReduce：Flink Connector

<dependency>
    <groupId>com.starrocks</groupId>
    <artifactId>flink-connector-starrocks</artifactId>
    <!-- for flink-1.11, flink-1.12 -->
    <version>x.x.x_flink-1.11</version>
    <!-- for flink-1.13 -->
    <version>x.x.x_flink-1.13</version>
</dependency>

• 方式一

  // -------- sink with raw json string stream --------
  fromElements(new String[]{
      "{\"score\": \"99\", \"name\": \"stephen\"}",
      "{\"score\": \"100\", \"name\": \"lebron\"}"
  }).addSink(
      StarRocksSink.sink(
          // the sink options
          StarRocksSinkOptions.builder()
              .withProperty("jdbc-url", "jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx")
              .withProperty("load-url", "fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port")
              .withProperty("username", "xxx")
              .withProperty("password", "xxx")
              .withProperty("table-name", "xxx")
              .withProperty("database-name", "xxx")
              .withProperty("sink.properties.format", "json")
              .withProperty("sink.properties.strip_outer_array", "true")
              .build()
      )
  );
  
  
  // -------- sink with stream transformation --------
  class RowData {
      public int score;
      public String name;
      public RowData(int score, String name) {
          ......
      }
  }
  fromElements(
      new RowData[]{
          new RowData(99, "stephen"),
          new RowData(100, "lebron")
      }
  ).addSink(
      StarRocksSink.sink(
          // the table structure
          TableSchema.builder()
              .field("score", DataTypes.INT())
              .field("name", DataTypes.VARCHAR(20))
              .build(),
          // the sink options
          StarRocksSinkOptions.builder()
              .withProperty("jdbc-url", "jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx")
              .withProperty("load-url", "fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port")
              .withProperty("username", "xxx")
              .withProperty("password", "xxx")
              .withProperty("table-name", "xxx")
              .withProperty("database-name", "xxx")
              .withProperty("sink.properties.column_separator", "\\x01")
              .withProperty("sink.properties.row_delimiter", "\\x02")
              .build(),
          // set the slots with streamRowData
          (slots, streamRowData) -> {
              slots[0] = streamRowData.score;
              slots[1] = streamRowData.name;
          }
      )
  );

• 方式二

  // create a table with `structure` and `properties`
  // Needed: Add `com.starrocks.connector.flink.table.StarRocksDynamicTableSinkFactory` to: `src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory`
  tEnv.executeSql(
      "CREATE TABLE USER_RESULT(" +
          "name VARCHAR," +
          "score BIGINT" +
      ") WITH ( " +
          "'connector' = 'starrocks'," +
          "'jdbc-url'='jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx'," +
          "'load-url'='fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port'," +
          "'database-name' = 'xxx'," +
          "'table-name' = 'xxx'," +
          "'username' = 'xxx'," +
          "'password' = 'xxx'," +
          "'sink.buffer-flush.max-rows' = '1000000'," +
          "'sink.buffer-flush.max-bytes' = '300000000'," +
          "'sink.buffer-flush.interval-ms' = '5000'," +
          "'sink.properties.column_separator' = '\\x01'," +
          "'sink.properties.row_delimiter' = '\\x02'," +
          "'sink.max-retries' = '3'" +
          "'sink.properties.*' = 'xxx'" + // stream load properties like `'sink.properties.columns' = 'k1, v1'`
      ")"
  );

• 如果有多組規則，則需要給每一組規則匹配database，table和flink-connector的配置。
  程式碼範例如下。

  [table-rule.1]
  # pattern to match databases for setting properties
  database = ^console_19321.*$
  # pattern to match tables for setting properties
  table = ^.*$
  
  ############################################
  ### flink sink configurations
  ### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
  ############################################
  flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
  flink.starrocks.load-url= 192.168.**.**:8030
  flink.starrocks.username=root
  flink.starrocks.password=
  flink.starrocks.sink.properties.column_separator=\x01
  flink.starrocks.sink.properties.row_delimiter=\x02
  flink.starrocks.sink.buffer-flush.interval-ms=15000
  
  [table-rule.2]
  # pattern to match databases for setting properties
  database = ^database2.*$
  # pattern to match tables for setting properties
  table = ^.*$
  
  ############################################
  ### flink sink configurations
  ### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
  ############################################
  flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
  flink.starrocks.load-url= 192.168.**.**:8030
  flink.starrocks.username=root
  flink.starrocks.password=
  # 如果匯入資料不方便選出合適的分隔字元，則可以考慮使用JSON格式，但是會有一定的效能損失，使用方法：用以下參數替換flink.starrocks.sink.properties.column_separator和flink.starrocks.sink.properties.row_delimiter參數。
  flink.starrocks.sink.properties.strip_outer_array=true
  flink.starrocks.sink.properties.format=json

  說明 Flink.starrocks.sink系列的參數，可以給不同的規則配置不同的匯入頻率等。
• 針對分庫分表的大表可以單獨配置一個規則。
  例如，有兩個資料庫edu_db_1和edu_db_2，每個資料庫下面分別有course_1、course_2兩張表，並且所有表的資料結構都是相同的，通過如以下配置把他們匯入StarRocks的一張表中進行分析。

  [table-rule.3]
  # pattern to match databases for setting properties
  database = ^edu_db_[0-9]*$
  # pattern to match tables for setting properties
  table = ^course_[0-9]*$
  
  ############################################
  ### flink sink configurations
  ### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
  ############################################
  flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
  flink.starrocks.load-url= 192.168.**.**:8030
  flink.starrocks.username=root
  flink.starrocks.password=
  flink.starrocks.sink.properties.column_separator=\x01
  flink.starrocks.sink.properties.row_delimiter=\x02
  flink.starrocks.sink.buffer-flush.interval-ms=5000

  配置後會自動產生一個多對一的匯入關係，在StarRocks預設產生的表名是course__auto_shard，也可以自行在產生的設定檔中修改。

• 如果在sql-client中命令列執行建表和同步任務，需要做對反斜線（\）字元進行轉義。
  程式碼範例如下。

  'sink.properties.column_separator' = '\\x01'
  'sink.properties.row_delimiter' = '\\x02'