配置 StarRocks 数据源

StarRocks 数据源，同步火山引擎 E-MapReduce（EMR）引擎中 StarRocks 集群数据库下的数据，为您提供离线、流式任务将数据写入 StarRocks 的单向通道能力。
本文为您介绍 DataSail 中 StarRocks 数据源配置、可视化配置能力说明。

1 支持的版本

支持采集 EMR-3.1.1 及以上 StarRocks 集群和 OLAP 服务中 1.1.0 及以上版本全托管 StarRocks 引擎中的数据。

2 使用前提

1. 子账号新建数据源时，需要有项目的管理员角色，方可以进行新建数据源操作。各角色对应权限说明，详见：管理成员
2. 访问 EMR-StarRocks 数据源，需先在项目控制台 ＞ 服务绑定 ＞ 引擎绑定中，绑定相应 EMR-StarRocks 集群。详见创建项目。
3. StarRocks 数据源配置时，EMR 集群对应的集群信息、数据库用户名密码需填写正确：

   注意

   填写的数据库用户名信息，必须拥有相应数据库表的读写权限，来保障任务数据能够被正常读取或写入 StarRocks 中。

   • 必须有账户密码，其中 root 账户无密码，不符合安全规范，数据源配置时无法使用。
   • 用户名密码获取方式，您可在 EMR 集群详情 ＞ 服务列表 ＞ StarRocks 服务名称 ＞ 服务参数 ＞ starrocks-env 参数文件下，看到 StarRocks 已经预置了一个账户和密码，推荐使用该账户/密码来配置 StarRocks 数据源。
     其余用户创建方式详见 CREATE USER。
     
4. EMR StarRocks 集群和独享集成资源组中的 VPC 必须一致。
   • 其 VPC 下的子网和安全组也尽可能保持一致。
   • 若 VPC 不一致时，则需要在 StarRocks 集群的安全组上，在入方向规则处，添加独享集成资源组子网的 IP 网段：
     1. 在 EMR StarRocks 集群详情界面，进入集群所在的安全组，并添加入方向规则。
        
     2. 在弹窗中，填写独享集成资源组子网的 IP 网段：
        

3 支持的字段类型与使用限制

支持大部分 StarRocks 类型，包括数值类型、字符串类型、日期类型：

4 数据同步任务开发

下文将为您介绍 StarRocks 数据集成同步任务的配置流程：

4.1 数据源注册

新建数据源操作详见配置数据源，下面为您介绍配置 StarRocks 数据源相关信息：

4.2 新建任务

StarRocks 数据源测试连通性成功后，进入到数据开发界面，开始新建 StarRocks 相关通道任务。
新建任务方式详见离线数据同步、流式数据同步。

4.3 可视化配置说明

任务创建成功后，您可根据实际场景，配置 StarRocks 批式读写、StarRocks 流式写等通道任务相关参数：

4.3.1 StarRocks 批式读

数据来源选择 StarRocks，并完成以下相关参数配置：
其中参数名称前带 * 的为必填参数，名称前未带 * 的为可选填参数。

4.3.2 StarRocks 批式写

数据目标端选择 StarRocks，并完成以下相关参数配置：
其中参数名称前带 * 的为必填参数，名称前未带 * 的为可选填参数。

4.3.3 StarRocks 流式写

流式数据写入 StarRocks 集成任务，数据目标端选择 StarRocks，并完成以下相关参数配置：
其中参数名称前带 * 的为必填参数，名称前未带 ＊ 的为可选填参数。

4.3.4 字段映射

数据来源和目标端配置完成后，需要指定来源和目标端的字段映射关系，根据字段映射关系，数据集成任务将源端字段中的数据，写入到目标端对应字段中。
您可通过以下三种方式操作字段映射关系：

• 自动添加：单击自动添加按钮，根据两端数据表信息，可以自动填充来源和目标的字段信息。
• 手动添加：单击手动添加按钮，可以手动编辑来源和目标的字段信息，可以逐个添加。

  说明

  来源端字段信息支持输入数据库函数和常量配置，单击手动添加按钮，在源表字段中输入需添加的值，并选择函数或常量类型，例如：

  • 函数：支持您输入 now()、current_timestamp() 等 StarRocks 数据库支持的函数。
  • 常量：您可自定义输入常量值，'123'、'${DATE}'、'${hour}' 等，输入值两侧需要加上英文单引号，支持结合平台时间变量与常量说明使用。
• 移动\删除字段：您也可以根据需要移动字段映射顺序或删除字段。

4.4 DSL 配置说明

StarRocks 数据源支持使用脚本模式（DSL）的方式进行配置。
在某些复杂场景下，或当数据源类型暂不支持可视化配置时，您可通过任务脚本的方式，按照统一的 Json 格式，编写 StarRocks Writer 参数脚本代码，来运行数据集成任务。

4.4.1 进入 DSL 模式

进入 DSL 模式操作流程，可详见 MySQL 数据源-4.4.1 进入DSL 模式。

4.4.2 StarRocks Writer

进入 DSL 模式编辑界面后，您可根据实际情况替换相应参数，StarRocks Writer 脚本示例如下：

• 未创建 StarRocks 数据源示例：

  // writer config
      "writer": {
           // [required] datasource type
          "type": "starrocks" ,
          // [optional] datasource id, set it if you have registered datasource
          "datasource_id": null,
          // [required] user parameter
          "parameter": {
               "password":"****************",
               "db_name":"sr_sink_test",
               "columns":[
                  {
                      "upperCaseName":"ID",
                      "name":"id",
                      "type":"BIGINT"
                  },
                  {
                      "upperCaseName":"NAME",
                      "name":"name",
                      "type":"STRING"
                  }
              ],
               "sink_write_mode":"STREAMING_UPSERT",
               "mysql_hosts":"172.16.1.2:9030",
               "fe_hosts":"172.16.1.2:8030",
               "user":"**********",
               "class":"com.bytedance.bitsail.connector.starrocks.sink.StarRocksSink",
               "table_name":"starrocks_sink_table_partation_dy_new",
               "pre_sql_list": ["delete from all_field_small_sink_1 where Id >= 0"],
               "post_sql_list": ["delete from all_field_small_sink_1 where Id >= 0"]
          }
      }
  

• 已创建 StarRocks 数据源示例：

  // writer config
      "writer": {
           // [required] datasource type
          "type": "starrocks" ,
          // [optional] datasource id, set it if you have registered datasource
          "datasource_id": 100,
          // [required] user parameter
          "parameter": {
               "columns":[
                  {
                      "upperCaseName":"ID",
                      "name":"id",
                      "type":"BIGINT"
                  },
                  {
                      "upperCaseName":"NAME",
                      "name":"name",
                      "type":"STRING"
                  }
              ],
               "sink_write_mode":"STREAMING_UPSERT",
               "class":"com.bytedance.bitsail.connector.starrocks.sink.StarRocksSink",
               "table_name":"starrocks_sink_table_partation_dy_new",
               "pre_sql_list": ["delete from all_field_small_sink_1 where Id >= 0"],
               "post_sql_list": ["delete from all_field_small_sink_1 where Id >= 0"]
          }
      }
  

• Writer 参数说明，其中参数名称前带＊的为必填参数，名称前未带＊的为可选填参数：

  参数名	参数说明
  *type	数据源类型，对于 StarRocks 类型，填写：starrocks
  *datasource_id	填写注册的 StarRocks 数据源 ID。可以在项目控制台 ＞ 数据源管理界面中查找。 目前 DSL 模式默认推荐不填写数据源 ID，这里可填写 null。 若通过数据源 ID 方式配置，下方的 user、password、db_name、mysql_hosts、fe_hosts 参数可不输入。
  *user	输入登录 StarRocks 集群数据库的用户名信息。
  *password	登录 StarRocks 数据库用户名对应的密码。
  *db_name	输入需要数据写入的 StarRocks 数据库名称。
  *table_name	填写数据源下所需数据写入的表名。
  *class	starrocks writer connector type，默认固定值： com.bytedance.bitsail.connector.starrocks.sink.StarRocksSink
  *columns	所配置的表中需要同步的列名集合，使用 JSON 的数组描述字段信息。 支持列裁剪：列可以挑选部分列进行导出。 支持列换序：列可以不按照表 Schema 信息顺序进行导出。 支持函数、常量形式添加列： 函数：StarRocks Reader 支持您输入 now()、current_timestamp() 等 StarRocks 数据库支持的函数。 常量：StarRocks Reader 支持您自定义输入常量值，如 '123'、'${DATE}'、'${hour}' 等，输入值两侧需要加上英文单引号，支持结合时间变量参数使用。
  *sink_write_mode	输入数据写入 StarRocks 的方式，当前仅支持 streaming_upsert 方式写入，参数值：STREAMING_UPSERT。
  *mysql_hosts	StarRocks 集群支持的 jdbc 节点信息，您可进入 EMR 控制台 ＞ StarRocks 集群详情 ＞ StarRocks 服务部署拓扑 ＞ Fe 组件界面，获取 Fe ip 信息，端口号通常默认为 9030。
  *fe_hosts	StarRocks 集群 Fe 节点信息，您可进入 EMR 控制台 ＞ StarRocks 集群详情 ＞ StarRocks 服务部署拓扑 ＞ Fe 组件界面，获取 Fe ip 信息，端口号通常默认为 8030。
  pre_sql_list	写入前准备语句：在执行数据集成任务前，率先执行的 SQL 语句。此语句通常是为了使任务重跑时支持幂等。 例如您可以通过填写写入前准备语句，清空表中的某些旧数据，清空完成后，再执行集成任务写入新的数据。如清空表中 p1,p2 分区：TRUNCATE TABLE table_name PARTITION(p1, p2); 说明 DSL 模式支持配置多条写入前准备语句，多条语句之间用英文逗号分隔。
  post_sql_list	写入后准备语句：执行数据同步任务后执行的 SQL 语句。例如数据写入完成后，插入某条特殊的数据，标志导入任务执行结束。 说明 DSL 模式支持配置多条写入后准备语句，多条语句之间用英文逗号分隔。