Read data from StarRocks using Flink connector

This topic describes how to use the source function of flink-connector-starrocks to read data from StarRocks.

If you need to use the sink function of flink-connector-starrocks to write data into StarRocks, see Flink connector in the Data Loading chapter.

Introduction

You can use the source function of flink-connector-starrocks to read data from StarRocks. Different from the Flink JDBC connector, flink-connector-starrocks can read data from multiple StarRocks backends (BEs) in parallel, which significantly improves data reading efficiency. Difference between the two connectors:

• flink-connector-starrocks: Flink first obtains the query plan from the frontend (FE), delivers the query plan as a parameter to BE nodes, and then obtains data results from BE nodes.

• Flink JDBC connector: Flink JDBC connector can only read data from the FE in a serial fashion. The data reading efficiency is low.

Procedure

Step 1: Install flink-connector-starrocks

1. Select a flink-connector-starrocks version based on your Flink version and download the JAR package of flink-connector-starrocks.

If you need to debug code, you can select the corresponding branch code and compile the code. 2. Place the downloaded or compiled JAR package in the lib directory of Flink. 3. Restart Flink.

Step 2: Use flink-connector-starrocks to read StarRocks data

The source function of flink-connector-starrocks cannot guarantee exactly-once semantics. If the reading task fails, you must repeat this step to create another reading task.

• If you use a Flink SQL client (recommended), you can read data from StarRocks by referring to the following command. For more information about the parameters in this command, see Parameter description.

-- Create a table in Flink based on the target StarRocks table and configure table properties (including information about flink-connector-starrocks, database, and table).

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'

);

-- Execute an SQL statement to read data from StarRocks.

select date_1, smallint_1 from flink_test where char_1 `<>` 'A' and int_1 = -126;

• Only some of the SQL statements can be used to read StarRocks data, such as

select ... from table_name where .... Aggregate functions except for COUNT are not supported.

• Predicate pushdown is supported. Predicates can be automatically pushed down when you execute SQL statements, such as the filter conditions in the preceding example.

char_1 < > 'A' and int_1 = -126 will be pushed down to the connector and converted into a statement suitable for querying StarRocks data. Additional configuration is not required.

• If you use Flink DataStream, you must add a dependency before you use flash-connector-starrocks to read StarRocks data.

Add the following dependency to the pom.xml file.

Replace x.x.x with the latest version number of flink-connector-starrocks. You can click version informationto obtain the latest version number.

<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>

Use flink-connector-starrocks to read data from StarRocks by referring to the following sample code. The following table describes the parameters in these commands.

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")

        .withProperty("cloumns", "char_1, date_1")        

        .withProperty("filters", "int_1 = 10")

        .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");

Parameter description

Data type mapping between Flink and StarRocks

The data type mapping in the following table applies only to reading StarRocks data from Flink. For the data type mapping for writing Flink data to StarRocks, see Flink connector in the Data Loading chapter.

What to do next

After data is read from StarRocks, you can use Flink WEBUI to observe the details of the reading task. For example, the Metrics page of Flink WEBUI displays the number of data rows that are read (totalScannedRows).