Read from and write to StarRocks - E-MapReduce: Central de Documentação do Alibaba Cloud

O StarRocks oferece um Spark Connector oficial para transferir dados entre o Spark e o StarRocks. Adicione as configurações necessárias no EMR Serverless Spark para conectar-se a uma instância do StarRocks. Este tópico descreve como ler e gravar dados no StarRocks com o EMR Serverless Spark.

Métodos de acesso

Acesse o StarRocks de duas formas, conforme sua necessidade:

Método 1: Configuração no nível da tarefa ou sessão

Neste método, defina as informações de conexão com o StarRocks — como URL JDBC, nome de usuário e senha — em cada tarefa ou sessão. Essa abordagem é adequada para os seguintes cenários:

Acesso temporário ou ad hoc aos dados
Tarefas diferentes precisam acessar clusters distintos do StarRocks
Controle de acesso refinado para cada tarefa

Método 2: Configuração centralizada com Data Catalog

Nota

Somente as seguintes versões de engine são suportadas: esr-4.8.0 ou posterior e esr-5.2.0 ou posterior.

Use o recurso Data Catalog do EMR Serverless Spark para adicionar um Data Catalog do StarRocks. Após a configuração, todos os jobs e sessões no workspace acessam os dados autorizados no Data Catalog por padrão, sem necessidade de configurar repetidamente cada tarefa. Para mais informações, consulte Gerencie data catalogs.

Esse método é recomendado nas situações abaixo:

Acesso frequente aos dados do StarRocks
Várias tarefas compartilham a mesma configuração de acesso ao StarRocks
Simplificação da configuração das tarefas e aumento da eficiência do desenvolvimento

Pré-requisitos

Crie um workspace do Serverless Spark. Para mais informações, consulte Crie a workspace.
Crie uma instância do EMR Serverless StarRocks. Para mais informações, consulte Crie an instance.

Limitações

A versão da engine do Serverless Spark deve ser esr-2.5.0, esr-3.1.0, esr-4.1.0 ou posterior.

Procedimento

Etapa 1: Obter e enviar o JAR do Spark Connector

Baixe o JAR do Spark Connector correspondente à sua versão de engine. Para mais informações, consulte Read data from StarRocks using Spark connector.

Nota
Para as versões de engine esr-4.8.0 ou posterior e esr-5.2.0 ou posterior, o JAR do Spark Connector já vem integrado. Pule esta etapa.

Neste exemplo, baixamos um JAR pré-compilado do Maven Central Repository.

Nota
O arquivo JAR do Connector segue o formato de nomenclatura starrocks-spark-connector-${spark_version}_${scala_version}-${connector_version}.jar. Por exemplo, se você usa a versão de engine esr-4.1.0 (Spark 3.5.2, Scala 2.12) e deseja utilizar a versão 1.1.2 do conector, selecione starrocks-spark-connector-3.5_2.12-1.1.2.jar.
Envie o JAR do Spark Connector baixado para o OSS. Para mais informações, consulte Simple upload.

Etapa 2: Adicionar uma conexão de rede

Obtenha as informações de rede.

Na página EMR Serverless StarRocks, acesse a página Instance Details da instância StarRocks desejada para obter as informações de Virtual Private Cloud (VPC) e vSwitch.
Crie uma Network Connection.
1. Na página EMR Serverless Spark, acesse a página Network Connection do seu workspace Spark e clique em Create Network Connection.
2. Na caixa de diálogo Create Network Connection, insira um Name, selecione a VPC e o vSwitch da instância StarRocks e clique em OK.
  
  Nota
  A Network Connection deve corresponder à instância StarRocks. Selecione um vSwitch na mesma VPC da instância StarRocks. Se não houver nenhum vSwitch disponível na zona de disponibilidade atual, clique em vSwitch para acessar o console VPC e criar um. Para mais informações, consulte VPCs and vSwitches.

Etapa 3: Criar banco de dados e tabela

Conecte-se à instância StarRocks. Para mais informações, consulte Connect to a StarRocks instance using EMR StarRocks Manager.
Na página SQL Editor Queries, clique em File ou no ícone à direita e, em seguida, clique em OK para criar um novo arquivo.

No novo arquivo, insira as instruções SQL abaixo e clique em Run.

CREATE DATABASE `testdb`;
CREATE TABLE `testdb`.`score_board`
(
    `id` int(11) NOT NULL COMMENT "",
    `name` varchar(65533) NULL DEFAULT "" COMMENT "",
    `score` int(11) NOT NULL DEFAULT "0" COMMENT ""
)
ENGINE=OLAP
PRIMARY KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`);

Etapa 4: Ler e gravar dados no StarRocks

Método 1: Sessões SQL e Notebook

Para mais informações sobre os tipos de sessão, consulte Session Manager.

Sessão SQL

Grave dados no StarRocks usando o EMR Serverless Spark.
1. Crie uma sessão SQL. Para mais informações, consulte Gerencie SQL sessions.
  Ao criar a sessão, selecione a versão da engine correspondente à versão do Spark Connector, escolha a conexão de rede criada na etapa anterior em Network Connection e adicione os seguintes parâmetros em Spark Configuration para carregar o Spark Connector.
```
spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar
```
  Substitua oss://<bucketname>/path/connector.jar pelo caminho do OSS do Spark Connector enviado na Etapa 1. Por exemplo, oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar.
2. Na página Development, crie uma tarefa SparkSQL. Em seguida, selecione a sessão SQL criada no canto superior direito.
  
  Para mais informações, consulte Develop a SparkSQL job.
3. Copie o código abaixo para a nova aba SparkSQL, substitua os valores dos placeholders conforme necessário e clique em Run.
```
CREATE TEMPORARY VIEW score_board
USING starrocks
OPTIONS
(
  "starrocks.table.identifier" = "testdb.score_board",
  "starrocks.fe.http.url" = "<fe_host>:<fe_http_port>",
  "starrocks.fe.jdbc.url" = "jdbc:mysql://<fe_host>:<fe_query_port>",
  "starrocks.user" = "<user>",
  "starrocks.password" = "<password>"
);
INSERT INTO `score_board` VALUES (1, "starrocks", 100), (2, "spark", 100);
```
  Descrição dos parâmetros:
  - <fe_host>: Endpoint interno ou público do nó FE na sua instância EMR Serverless StarRocks. Encontre essa informação na página Instance Details, na seção FE Details.
    - Se usar um endpoint interno, garanta que ambos os serviços estejam na mesma VPC.
    - Se usar um endpoint público, verifique se as regras do grupo de segurança permitem tráfego nas portas necessárias. Para mais informações, consulte Network access and security settings.
  - <fe_http_port>: Porta HTTP do nó FE na instância EMR Serverless StarRocks. O padrão é 8030. Consulte a página Instance Details, na seção FE Details.
  - <fe_query_port>: Porta de consulta do nó FE na instância EMR Serverless StarRocks. O padrão é 9030. Verifique na página Instance Details, dentro da seção FE Details.
  - <user>: Nome de usuário da instância Serverless StarRocks. Por padrão, há um usuário admin com privilégios administrativos. Também é possível adicionar novos usuários para conexão pela página User Management. Para detalhes sobre como adicionar usuários, consulte Manage Users and Data Authorization.
  - <password>: Senha correspondente ao <user>.
Consulte os dados gravados com o EMR Serverless Spark.

Neste exemplo, criamos uma visualização temporária chamada test_view na tarefa Spark SQL e a usamos para consultar dados de score_board. Copie o código abaixo para uma nova aba Spark SQL, selecione-o e clique em Run the selected.
```
CREATE TEMPORARY VIEW test_view
USING starrocks
OPTIONS
(
   "starrocks.table.identifier" = "testdb.score_board",
   "starrocks.fe.http.url" = "<fe_host>:<fe_http_port>",
   "starrocks.fe.jdbc.url" = "jdbc:mysql://<fe_host>:<fe_query_port>",
   "starrocks.user" = "<user>",
   "starrocks.password" = "<password>"
);
SELECT * FROM test_view;
```
Saída

A consulta retorna dois registros: (1, "starrocks", 100) e (2, "spark", 100).

Sessão Notebook

Grave dados no StarRocks usando o EMR Serverless Spark.
1. Crie uma sessão Notebook. Para mais informações, consulte Gerencie Notebook sessions.
  Ao criar a sessão, selecione a versão da engine correspondente à versão do Spark Connector, escolha a conexão de rede criada na etapa anterior em Network Connection e adicione os seguintes parâmetros em Spark Configuration para carregar o Spark Connector.
```
spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar
```
  Substitua oss://<bucketname>/path/connector.jar pelo caminho do OSS do Spark Connector enviado na Etapa 1. Por exemplo, oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar.
2. Na página Development, crie uma tarefa Interactive Development > Notebook e selecione a sessão Notebook criada no canto superior direito.
  
  Para mais informações, consulte Gerencie Notebook sessions.
3. Copie o código abaixo para uma célula Python no Notebook e clique em Run.
```
# Replace the placeholders with your EMR Serverless StarRocks configurations.
fe_host = "<fe_host>"
fe_http_port = "<fe_http_port>"
fe_query_port = "<fe_query_port>"
user = "<user>"
password = "<password>"
# Create a view.
create_table_sql = f"""
CREATE TEMPORARY VIEW score_board
USING starrocks
OPTIONS (
  "starrocks.table.identifier" = "testdb.score_board",
  "starrocks.fe.http.url" = "{fe_host}:{fe_http_port}",
  "starrocks.fe.jdbc.url" = "jdbc:mysql://{fe_host}:{fe_query_port}",
  "starrocks.user" = "{user}",
  "starrocks.password" = "{password}"
)
"""
spark.sql(create_table_sql)
# Insert data.
insert_data_sql = """
INSERT INTO `score_board` VALUES (1, "starrocks", 100), (2, "spark", 100)
"""
spark.sql(insert_data_sql)
```
  Descrição dos parâmetros:
  - <fe_host>: Endpoint interno ou público do nó FE na sua instância EMR Serverless StarRocks. Encontre essa informação na página Instance Details, na seção FE Details.
    - Se usar um endpoint interno, garanta que ambos os serviços estejam na mesma VPC.
    - Se usar um endpoint público, verifique se as regras do grupo de segurança permitem tráfego nas portas necessárias. Para mais informações, consulte Network access and security settings.
  - <fe_http_port>: Porta HTTP do nó FE na instância EMR Serverless StarRocks. O padrão é 8030. Consulte a página Instance Details, na seção FE Details.
  - <fe_query_port>: Porta de consulta do nó FE na instância EMR Serverless StarRocks. O padrão é 9030. Verifique na página Instance Details, dentro da seção FE Details.
  - <user>: Nome de usuário da instância Serverless StarRocks. Por padrão, há um usuário admin com privilégios administrativos. Também é possível adicionar novos usuários para conexão pela página User Management. Para detalhes sobre como adicionar usuários, consulte Manage Users and Data Authorization.
  - <password>: Senha correspondente ao <user>.

Consulte os dados gravados com o EMR Serverless Spark.

Em uma nova célula Python, crie uma visualização temporária chamada test_view para consultar a tabela score_board. Copie o código abaixo para a célula e clique no ícone de execução () para executá-lo.

# Create a view.
create_view_sql=f"""
CREATE TEMPORARY VIEW test_view
USING starrocks
OPTIONS (
  "starrocks.table.identifier" = "testdb.score_board",
  "starrocks.fe.http.url" = "{fe_host}:{fe_http_port}",
  "starrocks.fe.jdbc.url" = "jdbc:mysql://{fe_host}:{fe_query_port}",
  "starrocks.user" = "{user}",
  "starrocks.password" = "{password}"
)
"""
spark.sql(create_view_sql)
# Query the data.
query_sql="SELECT * FROM test_view"
result_df = spark.sql(query_sql)
result_df.show()

Saída

+---+---------+-----+
| id|     name|score|
+---+---------+-----+
|  2|    spark|  100|
|  1|starrocks|  100|
+---+---------+-----+

Método 2: Job em lote do Spark

Crie um job em lote do Spark.
1. Na página do EMR Serverless Spark, clique em Development no painel de navegação à esquerda.
2. Na aba Development, clique no ícone .
3. Na caixa de diálogo Create, insira um Name, defina o tipo como Application > SQL e clique em OK.
  
  Ajuste o tipo conforme sua necessidade. Este tópico usa SQL como exemplo. Para mais informações sobre tipos de job, consulte Develop an application.

Leia e grave dados no StarRocks usando o job em lote do Spark.

No canto superior direito da nova página de desenvolvimento de job, selecione uma fila.

Para mais informações sobre como adicionar uma fila, consulte Gerencie resource queues.

Na nova página de desenvolvimento de job, configure os parâmetros abaixo, mantenha os demais como padrão e clique em Run.

Parâmetro	Descrição
SQL File	Este exemplo utiliza o arquivo spark_sql_starrocks.sql, que contém as instruções SQL da sessão SQL. Antes de usar o arquivo, baixe-o, modifique as configurações conforme necessário e envie-o para a página Artifacts. Parâmetros do spark_sql_starrocks.sql Descrição dos parâmetros: `<fe_host>`: Endpoint interno ou público do nó FE na sua instância EMR Serverless StarRocks. Encontre essa informação na página Instance Details, na seção FE Details. Se usar um endpoint interno, garanta que ambos os serviços estejam na mesma VPC. Se usar um endpoint público, verifique se as regras do grupo de segurança permitem tráfego nas portas necessárias. Para mais informações, consulte Network access and security settings. `<fe_http_port>`: Porta HTTP do nó FE na instância EMR Serverless StarRocks. O padrão é 8030. Consulte a página Instance Details, na seção FE Details. `<fe_query_port>`: Porta de consulta do nó FE na instância EMR Serverless StarRocks. O padrão é 9030. Verifique na página Instance Details, dentro da seção FE Details. `<user>`: Nome de usuário da instância Serverless StarRocks. Por padrão, há um usuário admin com privilégios administrativos. Também é possível adicionar novos usuários para conexão pela página User Management. Para detalhes sobre como adicionar usuários, consulte Manage Users and Data Authorization. `<password>`: Senha correspondente ao `<user>`.
Engine Version	Selecione a versão da engine compatível com a versão do Spark Connector.
Normal Network Connection	Selecione a Network Connection criada na etapa anterior.
Spark Configuration	Adicione o seguinte parâmetro na seção Spark Configuration para carregar o Spark Connector. `spark.emr.serverless.user.defined.jars oss://<bucketname>/path/connector.jar` Substitua `oss://<bucketname>/path/connector.jar` pelo caminho do OSS do Spark Connector enviado na Etapa 1. Por exemplo, `oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar`.

Visualize as informações de log.
1. Na seção Execution Records na parte inferior, clique em Details na coluna Actions.
2. Clique na aba Log Exploration para visualizar os logs do job.
  
  Selecione Driver Log > Stdout para visualizar o arquivo stdout.log. Se o job for bem-sucedido, a saída do log conterá starrocks 100 e spark 100. Isso confirma que os dados foram gravados e lidos do StarRocks com sucesso.

Opções de configuração

Além das informações de conexão obrigatórias descritas acima, defina as opções abaixo conforme necessário.

Opção

Descrição

Como configurar

starrocks.warehouse

Warehouse usado para acessar o StarRocks. Se esta opção não for configurada, o warehouse padrão será utilizado.

Configure esta opção de uma das seguintes maneiras:

Na cláusula OPTIONS ao criar uma visualização temporária.
```
"starrocks.warehouse" = "<warehouse_name>"
```
Em um Data Catalog. Para mais informações, consulte Method 2: Centralized configuration with Data Catalog.
```
spark.sql.catalog.<catalog_name>.starrocks.warehouse=<warehouse_name>
```

Referências

Para mais detalhes, consulte a documentação oficial do StarRocks:

Read data from StarRocks using Spark connector | StarRocks