连接

Flink 集群启动成功后，用户可通过 FlinkSQL 客户端访问 SequoiaDB 巨杉数据库。

配置读取并发度

从 SequoiaDB 读取数据时，用户需在 FlinkSQL 客户端中配置读取并发度，以保证性能。配置读取并发度时，建议取值与 SequoiaDB 集群中复制组的数量成比例。如集群中存在 6 个复制组，建议将读取并发度配置为 6、12 或 24。

1. 切换至 Flink 安装目录，并启动 FlinkSQL 客户端

   $ bin/sql-client.sh

2. 配置读取并发度为 12

   Flink SQL> SET 'parallelism.default'='12';

创建映射表

Flink 与 SequoiaDB 通过映射表进行数据交互。

1. 切换至 Flink 安装目录，并启动 FlinkSQL 客户端

   $ bin/sql-client.sh

2. 在 Flink 中创建映射表 employee，映射对象为协调节点 sdbserver1:11810 所在集群的集合 sample.employee

   Flink SQL> CREATE TABLE employee (id INT PRIMARY KEY, name STRING, age INT)
   WITH (
   'connector' = 'sequoiadb',
   'hosts' = 'sdbserver1:11810',
   'collectionspace' = 'sample',
   'collection' = 'employee'
   );

   Note:

   如果所映射的集合空间或集合不存在于目标集群中，连接器会在数据写入时自动创建。当映射表存在主键时，连接器会在对应集合中创建名为“primarykey”的唯一索引，该索引的索引键为主键所包含的字段。

自定义表配置

Flink 创建映射表语句的格式如下：

CREATE TABLE [IF NOT EXISTS] [catalog_name.][db_name.]table_name (<column1, column2, ...>)
WITH(
'option1'='value1', 
'option2'='value2'
...
);

创建成功的映射表，在被使用之前都不会与 SequoiaDB 建立连接，因此不会检测参数 option 所指定配置的正确性。

通过参数 option 可以配置映射表的属性，支持的配置项如下：

保证精确一次性

FlinkSQL 连接器通过 Checkpoint 机制保证数据写出的精确一次性。当用户将表配置 overwrite 设置为 true 时，需根据实际场景配置 Flink 集群中与 StateBackend 有关的参数，以保证数据处理的精确一次。

下述以使用 RocksDB 增量存储 Checkpoint 为例，演示部分参数的配置方式，用户需根据实际情况调整配置。

1. 配置 Checkpoint 间隔，以启用 Checkpoint 机制

   execution.checkpointing.inteval: 5 min

2. 配置 StateBackend 相关参数

   state.backend: rocksdb
   state.backend.incremental: true
   state.checkpoints.dir: hdfs:///flink-checkpoint        # location to store checkpoints
   ···

3. 确保已正确配置 RocksDB 中的相关参数