环境准备

创建TOS桶

1. 登录TOS控制台，创建TOS桶，具体创建方式可参考文档：创建TOS桶；
2. 基于您创建的TOS桶创建您所需的目录，该目录作为Starrocks实例的默认存储目录。

Starrocks数据准备

1. 登录EMR Serverless StarRocks控制台，创建StarRocks实例，具体创建方式可参考火山官网文档：创建StarRocks实例
2. 连接实例后，执行下列SQL进行数据库&表准备，连接实例方式可见官网文档：连接实例。

CREATE DATABASE if not EXISTS online_database;
USE online_database;
CREATE TABLE log_v1 (
    uid int,
    name varchar(64),
    age int, 
    phone varchar(16),
    last_login datetime,
    credits double
)
DUPLICATE KEY(uid, name);

CREATE TABLE log_v2 (
    uid int,
    name varchar(64),
    age int, 
    phone varchar(16),
    last_login datetime,
    credits double
)
DUPLICATE KEY(uid, name);

3. 数据准备完成后，您可进行数据查看，数据展示内容如下则表示数据准备完成：

Serverless Spark数据准备

1. 创建资源队列：登录EMR Serverless控制台创建Serverless Spark队列，根据需求创建按量付费或者包年包月队列，当队列状态变为运行中时，您可开始下一步操作；
2. 创建数据目录：登录湖仓一体分析服务创建自定义数据目录(Catalog)，并为其设置TOS存储位置；

3. 创建数据库：登录EMR Serverless Spark作业编辑页面，使用默认数据目录hive，创建数据库实例ss_w2r_sr，使用sql如下：

CREATE DATABASE IF NOT EXISTS ss_w2r_sr;

4. 数据表：登录EMR Serverless Spark作业编辑页面，使用默认数据目录hive，数据库实例ss_w2r_sr创建两张数据表log_v1和log_v2。其中，location为数据库实例TOS地址+数据表名称，数据库实例地址可以通过湖仓一体分析服务查询。

CREATE TABLE if not exists ss_w2r_sr.log_v1(
  `uid` int, 
  `name` string, 
  `age` int, 
  `phone` string, 
  `last_login` TIMESTAMP, 
  `credits` double)
PARTITIONED BY ( 
  `date` string)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
location 'tos://{Database TOS storage address}/log_v1';

CREATE TABLE if not exists ss_w2r_sr.log_v2(
  `uid` int, 
  `name` string, 
  `age` int, 
  `phone` string, 
  `last_login` TIMESTAMP, 
  `credits` double)
PARTITIONED BY ( 
  `date` string)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
location 'tos://{Database TOS storage address}/log_v2';

5. 配置安全组：登陆火山引擎，创建安全组实例，在私有网络列表中选择Starrocks实例使用的VPC，并按下图进行入规则设置，完成安全组实例的创建：

执行步骤

1. 通过Serverless Spark对Starrocks数据表进行读写

登录EMR Serverless Spark作业编辑页面，输入下列SQL代码块中示例的Spark SQL语句，点击作业编辑框右下角的“运行”按钮，执行SQL作业。

-- starrocks fe url
set spark.sql.catalog.starrocks.fe.http.url=http://{local ip address}:8030;
set spark.sql.catalog.starrocks.fe.jdbc.url=jdbc:mysql://{local ip address}:9030;
-- user name with admin right in starrocks instance
set spark.sql.catalog.starrocks.user={user name};
-- user password
set spark.sql.catalog.starrocks.password={user password};
-- starrocks instance TOS endpoint, such as https://tos-s3-cn-beijing.ivolces.com
set spark.sql.catalog.starrocks.fs.s3a.endpoint={endpoint};
-- starrocks instance TOS region, such as cn-beijing
set spark.sql.catalog.starrocks.fs.s3a.endpoint.region={region name};
set spark.sql.catalog.starrocks.fs.s3a.access.key={access key};
set spark.sql.catalog.starrocks.fs.s3a.secret.key={secret key};

set tqs.query.engine.type = sparkcli;
set las.cross.vpc.access.enabled = true;
set las.cross.vpc.vpc.id = {vpc id};
set las.cross.vpc.subnet.id = {subnet id};
set las.cross.vpc.security.group.id = {security group id};
set las.cross.vpc.accountId = {volc account id};

set spark.sql.catalog.starrocks.batch.size=8092;
set spark.sql.catalog.starrocks.writer.mode=BYPASS;
set spark.sql.catalog.starrocks.reader.mode=BYPASS;
set spark.sql.catalog.starrocks.fs.s3a.connection.ssl.enabled=true;
set spark.sql.catalog.starrocks.fs.s3a.path.style.access=false;
set spark.sql.catalog.starrocks.fs.s3a.retry.limit=27;
set spark.sql.catalog.starrocks=com.starrocks.connector.spark.catalog.StarRocksCatalog;
set spark.sql.storeAssignmentPolicy = ansi;
-- depend jar address, please refer to "Dependency JAR addresses" for jar addresses of online regions
set las.spark.jar.depend.jars = [{"fileName":"{dependency jar address}"}];

set spark.shuffle.manager = sort;
set spark.dynamicAllocation.enabled = false;
set spark.sql.serverless.lf.auth.enabled = false;
set spark.sql.extensions=com.starrocks.connector.spark.StarRocksExtensions;

use starrocks;
insert into table online_database.log_v2
select * from online_database.log_v1;
select * from online_database.log_v2;

2. 通过Serverless Spark读Starrocks数据表，写Hive数据表

登录EMR Serverless Spark作业编辑页面，输入下列SQL代码块中示例的Spark SQL语句，点击作业编辑框右下角的“运行”按钮，执行SQL作业。

-- starrocks fe url
set spark.sql.catalog.starrocks.fe.http.url=http://{local ip address}:8030;
set spark.sql.catalog.starrocks.fe.jdbc.url=jdbc:mysql://{local ip address}:9030;
-- user name with admin right in starrocks instance
set spark.sql.catalog.starrocks.user={user name};
-- user password
set spark.sql.catalog.starrocks.password={user password};
-- starrocks instance TOS endpoint, such as https://tos-s3-cn-beijing.ivolces.com
set spark.sql.catalog.starrocks.fs.s3a.endpoint={endpoint};
-- starrocks instance TOS region, such as cn-beijing
set spark.sql.catalog.starrocks.fs.s3a.endpoint.region={region name};
set spark.sql.catalog.starrocks.fs.s3a.access.key={access key};
set spark.sql.catalog.starrocks.fs.s3a.secret.key={secret key};

set tqs.query.engine.type = sparkcli;
set las.cross.vpc.access.enabled = true;
set las.cross.vpc.vpc.id = {vpc id};
set las.cross.vpc.subnet.id = {subnet id};
set las.cross.vpc.security.group.id = {security group id};
set las.cross.vpc.accountId = {volc account id};

set spark.sql.catalog.starrocks.batch.size=8092;
set spark.sql.catalog.starrocks.writer.mode=BYPASS;
set spark.sql.catalog.starrocks.reader.mode=BYPASS;
set spark.sql.catalog.starrocks.fs.s3a.connection.ssl.enabled=true;
set spark.sql.catalog.starrocks.fs.s3a.path.style.access=false;
set spark.sql.catalog.starrocks.fs.s3a.retry.limit=27;
set spark.sql.catalog.starrocks=com.starrocks.connector.spark.catalog.StarRocksCatalog;
set spark.sql.storeAssignmentPolicy = ansi;
-- depend jar address, please refer to "Dependency JAR addresses" for jar addresses of online regions
set las.spark.jar.depend.jars = [{"fileName":"{dependency jar address}"}];

set spark.shuffle.manager = sort;
set spark.dynamicAllocation.enabled = false;
set spark.sql.serverless.lf.auth.enabled = false;
set spark.sql.extensions=com.starrocks.connector.spark.StarRocksExtensions;
                                                     
use starrocks;
WITH temp AS (
    select uid, name, age, phone, last_login, credits from online_database.log_v1
)
insert overwrite table spark_catalog.ss_w2r_sr.log_v1 partition(date)
select uid + cast(rand() * 900 + 10 as int), CONCAT(name,'_1'), age, phone, last_login, credits, '20240601' from temp;
select * from spark_catalog.ss_w2r_sr.log_v1;

3. 通过Serverless Spark读Hive数据表，写入Starrocks数据表

登录EMR Serverless Spark作业编辑页面，输入下列SQL代码块中示例的Spark SQL语句，点击作业编辑框右下角的“运行”按钮，执行SQL作业。

-- starrocks fe url
set spark.sql.catalog.starrocks.fe.http.url=http://{local ip address}:8030;
set spark.sql.catalog.starrocks.fe.jdbc.url=jdbc:mysql://{local ip address}:9030;
-- user name with admin right in starrocks instance
set spark.sql.catalog.starrocks.user={user name};
-- user password
set spark.sql.catalog.starrocks.password={user password};
-- starrocks instance TOS endpoint, such as https://tos-s3-cn-beijing.ivolces.com
set spark.sql.catalog.starrocks.fs.s3a.endpoint={endpoint};
-- starrocks instance TOS region, such as cn-beijing
set spark.sql.catalog.starrocks.fs.s3a.endpoint.region={region name};
set spark.sql.catalog.starrocks.fs.s3a.access.key={access key};
set spark.sql.catalog.starrocks.fs.s3a.secret.key={secret key};

set tqs.query.engine.type = sparkcli;
set las.cross.vpc.access.enabled = true;
set las.cross.vpc.vpc.id = {vpc id};
set las.cross.vpc.subnet.id = {subnet id};
set las.cross.vpc.security.group.id = {security group id};
set las.cross.vpc.accountId = {volc account id};

set tqs.query.engine.type = sparkcli;
set las.cross.vpc.access.enabled = true;
set las.cross.vpc.vpc.id = {vpc id};
set las.cross.vpc.subnet.id = {subnet id};
set las.cross.vpc.security.group.id = {security group id};
set las.cross.vpc.accountId = {volc account id};

set spark.sql.catalog.starrocks.batch.size=8092;
set spark.sql.catalog.starrocks.writer.mode=BYPASS;
set spark.sql.catalog.starrocks.reader.mode=BYPASS;
set spark.sql.catalog.starrocks.fs.s3a.connection.ssl.enabled=true;
set spark.sql.catalog.starrocks.fs.s3a.path.style.access=false;
set spark.sql.catalog.starrocks.fs.s3a.retry.limit=27;
set spark.sql.catalog.starrocks=com.starrocks.connector.spark.catalog.StarRocksCatalog;
set spark.sql.storeAssignmentPolicy = ansi;
-- depend jar address, please refer to "Dependency JAR addresses" for jar addresses of online regions
set las.spark.jar.depend.jars = [{"fileName":"{dependency jar address}"}];

set spark.shuffle.manager = sort;
set spark.dynamicAllocation.enabled = false;
set spark.sql.serverless.lf.auth.enabled = false;
set spark.sql.extensions=com.starrocks.connector.spark.StarRocksExtensions;
        
use starrocks;
WITH temp AS (
  select uid, name, age, phone, last_login, credits from spark_catalog.ss_w2r_sr.log_v1
)
insert into table online_database.log_v1 
select * from temp;
select * from online_database.log_v1;

Dependency JAR addresses