Flink Connector

本文介绍 Flink 如何通过 flink-connector-starrocks 的 source 功能读取 StarRocks 数据。

如果 Flink 需要通过 flink-connector-starrocks 的 sink 功能，将数据写入至 StarRocks，请参见数据导入章节的 Flink connector。

功能简介

Flink 可以通过 flink-connector-starrocks 的 source 功能读取 StarRocks 的数据。相较于 Flink 官方提供的 Flink JDBC connector，flink-connector-starrocks 的 source 功能具备并行读取 StarRocks 的 BE 节点数据的能力，大大提高了数据读取效率。以下是两种连接器的实现方案对比。

• flink-connector-starrocks 的实现方案：Flink 先从 FE 节点获取查询计划（Query Plan），Flink 再将获取到的查询计划作为参数，下发至 BE 节点，然后获取 BE 节点返回的数据。

  

• Flink JDBC connector 的实现方案：Flink JDBC connector 仅能从 FE 单点上串行读取数据，数据读取效率较低。

  

操作步骤

步骤一：准备flink-connector-starrocks

1. 根据 Flink 的版本，选择对应版本。下载 JAR 包 flink-connector-starrocks。
2. 如需调试代码，可选择对应分支代码自行编译
3. 将下载或者编译的 JAR 包放在 Flink 的 lib 目录中。
4. 重启 Flink。

步骤二：调用 flink-connector-starrocks ，读取 StarRocks 数据

flink-connector-starrocks 的 source 功能暂时无法保证 exactly-once 语义。如果读取任务失败，您需要重复本步骤，再次创建读取任务。

• 如您使用 Flink SQL 客户端（推荐），则需要参考如下命令，调用 flink-connector-starrocks，读取 StarRocks 的数据。相关参数说明，请参见参数说明。

  -- 根据 StarRocks 的表，创建表和配置属性（包括 flink-connector-starrocks 和库表的信息）。
  CREATE TABLE flink_test (
      date_1 DATE,
      datetime_1 TIMESTAMP(6),
      char_1 CHAR(20),
      varchar_1 VARCHAR,
      boolean_1 BOOLEAN,
      tinyint_1 TINYINT,
      smallint_1 SMALLINT,
      int_1 INT,
      bigint_1 BIGINT,
      largeint_1 STRING,
      float_1 FLOAT,
      double_1 DOUBLE,FLI
      decimal_1 DECIMAL(27,9)
  ) WITH (
     'connector'='starrocks',
     'scan-url'='192.168.xxx.xxx:8030,192.168.xxx.xxx:8030',
     'jdbc-url'='jdbc:mysql://192.168.xxx.xxx:9030',
     'username'='root',
     'password'='xxxxxx',
     'database-name'='flink_test',
     'table-name'='flink_test'
  );
  -- 使用 SQL 语句读取 StarRocks 数据。
  select date_1, smallint_1 from flink_test where char_1 `<>` 'A' and int_1 = -126;

  • 仅支持使用部分 SQL 语句读取 StarRocks 数据，如select ... from table_name where ...。暂不支持除 COUNT 外的聚合函数。
  • 支持谓词下推。使用 SQL 语句时，支持自动进行谓词下推，比如上述例子中的过滤条件 char_1 <> 'A' and int_1 = -126，会下推到 connector 中转换成适用于 StarRocks 的语句进行查询，不需要额外配置。
  • 不支持limit语句
  • 在任务失败的情况下，由于暂时没有 checkpoint 机制，无法保证数据一致性

• 如您使用 Flink DataStream ，则需要先添加依赖，然后调用 flink-connector-starrocks，读取 StarRocks 的数据。

1. 在 pom.xml 文件中添加如下依赖。

   x.x.x需要替换为 flink-connector-starrocks 的最新版本号，您可以单击版本信息获取。

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

2. 参考如下示例代码，调用 flink-connector-starrocks，读取 StarRocks 的数据。相关参数说明，请参见参数说明。

import com.starrocks.connector.flink.StarRocksSource;
import com.starrocks.connector.flink.table.source.StarRocksSourceOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;

public class StarRocksSourceApp {
    public static void main(String[] args) {
        StarRocksSourceOptions options = StarRocksSourceOptions.builder()
               .withProperty("scan-url", "192.168.xxx.xxx:8030,192.168.xxx.xxx:8030")
               .withProperty("jdbc-url", "jdbc:mysql://192.168.xxx.xxx:9030")
               .withProperty("username", "root")
               .withProperty("password", "xxxxxx")
               .withProperty("table-name", "flink_test")
               .withProperty("database-name", "test")
               .build();
        TableSchema tableSchema = TableSchema.builder()
               .field("date_1", DataTypes.DATE())
               .field("datetime_1", DataTypes.TIMESTAMP(6))
               .field("char_1", DataTypes.CHAR(20))
               .field("varchar_1", DataTypes.STRING())
               .field("boolean_1", DataTypes.BOOLEAN())
               .field("tinyint_1", DataTypes.TINYINT())
               .field("smallint_1", DataTypes.SMALLINT())
               .field("int_1", DataTypes.INT())
               .field("bigint_1", DataTypes.BIGINT())
               .field("largeint_1", DataTypes.STRING())
               .field("float_1", DataTypes.FLOAT())
               .field("double_1", DataTypes.DOUBLE())
               .field("decimal_1", DataTypes.DECIMAL(27, 9))
               .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(StarRocksSource.source(options, tableSchema)).setParallelism(5).print();
        env.execute("StarRocks flink source");
    }

}

参数说明

Streaming中特有参数

Flink 与 StarRocks 的数据类型映射关系

该数据类型映射关系仅适用于 Flink 读取 StarRocks 数据。如需要查看 Flink 将数据写入至 StarRocks 的数据类型映射关系，请参见数据导入章节的 Flink connector。

后续步骤

Flink 成功读取 StarRocks 数据后，您可以使用 Flink 官方的 Flink WEBUI 界面观察读取任务。比如 Flink WEBUI 的 Metrics 页面会显示成功读取的数据行数（totalScannedRows）。