开源大数据平台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'