E-MapReduce:Manage Spark Thrift Server sessions

Conectar ao Spark Thrift Server usando Python

1. Execute o seguinte comando para instalar os pacotes PyHive e Thrift.

   pip install pyhive thrift

2. Escreva um script Python para conectar ao Spark Thrift Server.

   O script a seguir conecta ao Hive e lista todos os bancos de dados. Escolha um método de conexão de acordo com seu ambiente de rede.

   Connect using a public endpoint

   from pyhive import hive
   
   if __name__ == '__main__':
       # Replace <endpoint>, <username>, and <token> with your actual information.
       cursor = hive.connect('<endpoint>', port=443, scheme='https', username='<username>', password='<token>').cursor()
       cursor.execute('show databases')
       print(cursor.fetchall())
       cursor.close()

   Connect using an internal same-region endpoint

   from pyhive import hive
   
   if __name__ == '__main__':
       # Replace <endpoint>, <username>, and <token> with your actual information.
       cursor = hive.connect('<endpoint>', port=80, scheme='http', username='<username>', password='<token>').cursor()
       cursor.execute('show databases')
       print(cursor.fetchall())
       cursor.close()

Conectar ao Spark Thrift Server usando Java

1. Adicione as seguintes dependências Maven ao arquivo pom.xml.

   <dependencies>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
               <version>3.0.0</version>
           </dependency>
           <dependency>
               <groupId>org.apache.hive</groupId>
               <artifactId>hive-jdbc</artifactId>
               <version>2.1.0</version>
           </dependency>
       </dependencies>
   

   Nota

   A versão integrada do Hive no Serverless Spark é 2.x. Portanto, apenas o hive-jdbc 2.x é compatível.

2. Escreva o código Java para conectar ao Spark Thrift Server.

   O código a seguir conecta ao Spark Thrift Server e consulta a lista de bancos de dados.

   Connect using a public endpoint

   import java.sql.Connection;
   import java.sql.DriverManager;
   import java.sql.ResultSet;
   import java.sql.ResultSetMetaData;
   import org.apache.hive.jdbc.HiveStatement;
   
   public class Main {
       public static void main(String[] args) throws Exception {
           String url = "jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token>";
           Class.forName("org.apache.hive.jdbc.HiveDriver");
           Connection conn = DriverManager.getConnection(url);
           HiveStatement stmt = (HiveStatement) conn.createStatement();
   
           String sql = "show databases";
           System.out.println("Running " + sql);
           ResultSet res = stmt.executeQuery(sql);
   
           ResultSetMetaData md = res.getMetaData();
           String[] columns = new String[md.getColumnCount()];
           for (int i = 0; i < columns.length; i++) {
               columns[i] = md.getColumnName(i + 1);
           }
           while (res.next()) {
               System.out.print("Row " + res.getRow() + "=[");
               for (int i = 0; i < columns.length; i++) {
                   if (i != 0) {
                       System.out.print(", ");
                   }
                   System.out.print(columns[i] + "='" + res.getObject(i + 1) + "'");
               }
               System.out.println(")]");
           }
           
           conn.close();
       }
   }

   Connect using an internal same-region endpoint

   import java.sql.Connection;
   import java.sql.DriverManager;
   import java.sql.ResultSet;
   import java.sql.ResultSetMetaData;
   import org.apache.hive.jdbc.HiveStatement;
   
   public class Main {
       public static void main(String[] args) throws Exception {
           String url = "jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>";
           Class.forName("org.apache.hive.jdbc.HiveDriver");
           Connection conn = DriverManager.getConnection(url);
           HiveStatement stmt = (HiveStatement) conn.createStatement();
   
           String sql = "show databases";
           System.out.println("Running " + sql);
           ResultSet res = stmt.executeQuery(sql);
   
           ResultSetMetaData md = res.getMetaData();
           String[] columns = new String[md.getColumnCount()];
           for (int i = 0; i < columns.length; i++) {
               columns[i] = md.getColumnName(i + 1);
           }
           while (res.next()) {
               System.out.print("Row " + res.getRow() + "=[");
               for (int i = 0; i < columns.length; i++) {
                   if (i != 0) {
                       System.out.print(", ");
                   }
                   System.out.print(columns[i] + "='" + res.getObject(i + 1) + "'");
               }
               System.out.println(")]");
           }
           
           conn.close();
       }
   }

Conectar ao Spark Thrift Server usando Spark Beeline

• Se você estiver usando um cluster autogerenciado, acesse o diretório bin do Spark e utilize o Beeline para conectar ao Spark Thrift Server.

  Connect using a public endpoint

  cd /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3/bin/
  
  ./beeline -u "jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token>"

  Connect using an internal same-region endpoint

  cd /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3/bin/
  
  ./beeline -u "jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>"

  Nota

  O caminho /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3 no código é um exemplo do caminho de instalação do Spark em um cluster EMR on ECS. Ajuste o caminho de acordo com o caminho real de instalação do Spark no seu cliente. Se você não souber o caminho de instalação do Spark, execute o comando env | grep SPARK_HOME para localizá-lo.

• Se você estiver usando um cluster EMR on ECS, utilize diretamente o cliente Spark Beeline para conectar ao Spark Thrift Server.

  Connect using a public endpoint

  spark-beeline -u "jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token>"

  Connect using an internal same-region endpoint

  spark-beeline -u "jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>"

Se o erro a seguir ocorrer ao usar o Hive Beeline para conectar ao Serverless Spark Thrift Server, a causa geralmente é uma incompatibilidade entre a versão do Hive Beeline e o Spark Thrift Server. Para resolver o problema, use uma versão 2.x do Beeline para Hive.

24/08/22 15:09:11 [main]: ERROR jdbc.HiveConnection: Error opening session
org.apache.thrift.transport.TTransportException: HTTP Response code: 404

Configure o Apache Superset para conectar ao Spark Thrift Server

O Apache Superset é uma plataforma de exploração e visualização de dados que suporta uma ampla variedade de tipos de gráficos. Para mais informações, consulte a documentação do Superset.

1. Instale as dependências.

   Verifique se a versão 0.20.0 do pacote thrift está instalada. Caso contrário, execute o comando a seguir para instalá-la.

   pip install thrift==0.20.0

2. Inicie o Superset e acesse a interface do Superset.

   Para mais informações sobre como iniciar o Superset, consulte a documentação do Superset.

3. No canto superior direito da página, clique em DATABASE para acessar a página Connect a database.

4. Na página Connect a database, selecione Apache Spark SQL.

   

5. Insira a string de conexão e configure os parâmetros da fonte de dados.

   Connect using a public endpoint

   hive+https://<username>:<token>@<endpoint>:443/<db_name>

   Connect using an internal same-region endpoint

   hive+http://<username>:<token>@<endpoint>:80/<db_name>

6. Clique em FINISH para confirmar a conexão e a validação.

Configure o Hue para conectar ao Spark Thrift Server

O Hue é uma interface web de código aberto para interagir com o ecossistema Hadoop. Para mais informações, consulte a documentação do Hue.

1. Instale as dependências.

   Verifique se a versão 0.20.0 do pacote thrift está instalada. Caso contrário, execute o comando a seguir para instalá-la.

   pip install thrift==0.20.0

2. Adicione a string de conexão do Spark SQL ao arquivo de configuração do Hue.

   Localize o arquivo de configuração do Hue (normalmente em /etc/hue/hue.conf) e adicione o seguinte conteúdo ao arquivo.

   Connect using a public endpoint

   [[[sparksql]]]
        name = Spark Sql
        interface=sqlalchemy
        options='{"url": "hive+https://<username>:<token>@<endpoint>:443/"}'

   Connect using an internal same-region endpoint

   [[[sparksql]]]
        name = Spark Sql
        interface=sqlalchemy
        options='{"url": "hive+http://<username>:<token>@<endpoint>:80/"}'

3. Reinicie o Hue.

   Após modificar a configuração, execute o comando a seguir para reiniciar o service do Hue e aplicar as alterações.

   sudo service hue restart

4. Verifique a conexão.

   Após reiniciar com sucesso, acesse a interface do Hue e localize a opção Spark SQL. Se a configuração estiver correta, você poderá conectar ao Spark Thrift Server e executar consultas SQL.

   

Conectar ao Spark Thrift Server usando DataGrip

O DataGrip é uma ferramenta de gerenciamento de banco de dados para consultar, criar e gerenciar bancos de dados em ambientes locais, remotos ou cloud. Para mais informações, acesse o site do DataGrip.

1. Instale o DataGrip. Para mais informações, consulte Install DataGrip.

   A versão do DataGrip utilizada neste exemplo é a 2025.1.2.

2. Abra o cliente DataGrip. A interface do DataGrip será exibida.

3. Crie um projeto.

   1. Clique em e escolha New.

      

   2. Na caixa de diálogo New Project, insira um nome para o projeto, como Spark, e clique em OK.

4. Clique no ícone na barra de menus do Database Explorer. Selecione Data Source > Other > Apache Spark.

   

5. Na caixa de diálogo Data Sources and Drivers, configure os seguintes parâmetros.

   

   Tab	Parameter	Descrição
   General	Name	Um nome personalizado para a conexão. Por exemplo, spark_thrift_server.
   	Authentication	Selecione um método de autenticação. Neste tópico, a opção No auth está selecionada. Em ambientes de produção, selecione User & Password para garantir que apenas usuários autorizados possam enviar jobs SQL, aumentando a segurança do sistema.
   	Driver	Clique em Apache Spark e, em seguida, clique em Go to Driver para confirmar que a versão do driver é ver. 1.2.2. Nota Como a versão atual do engine do Serverless Spark é 3.x, selecione a versão 1.2.2 do driver para garantir estabilidade e compatibilidade de recursos.
   	URL	A URL de conexão com o Spark Thrift Server. Escolha um método de conexão de acordo com seu ambiente de rede. Conexão via endpoint público jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token> Conexão via endpoint interno na mesma região jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>
   Options	Run keep-alive query	Opcional. Ative esta opção para evitar a desconexão automática por tempo de inatividade.

6. Clique em Test Connection para verificar a conexão.

   

7. Clique em OK para concluir a configuração.

8. Agora você pode gerenciar o Spark Thrift Server usando o DataGrip.

   Após conectar o DataGrip ao Spark Thrift Server, você poderá realizar o desenvolvimento de dados. Para mais informações, consulte a documentação de ajuda do DataGrip.

   Por exemplo, na conexão criada, clique com o botão direito na tabela de destino, escolha New e, no editor SQL que se abre, escreva e execute um script SQL para visualizar os dados da tabela.

   

Conectar ao Spark Thrift Server usando Redash

O Redash é uma ferramenta de BI de código aberto para consultas em banco de dados via web e visualização de dados. Para mais informações, consulte a documentação do Redash.

1. Instale o Redash. Para mais informações, consulte a documentação oficial do Redash.

2. Instale as dependências.

   Verifique se a versão 0.20.0 do pacote thrift está instalada. Caso contrário, execute o comando a seguir para instalá-la.

   pip install thrift==0.20.0

3. Faça login no Redash.

4. No painel de navegação à esquerda, clique em Settings e, na aba Data Sources, clique em +New Data Source.

5. Na caixa de diálogo exibida, configure os parâmetros e clique em Create.

   

   Parâmetro		Descrição
   Type Selection		O tipo de fonte de dados. Na caixa de pesquisa, localize e selecione Hive (HTTP).
   Configuration	Name	O nome da fonte de dados. É possível personalizar o nome.
   	Host	O endpoint do Spark Thrift Server. Obtenha o Endpoint(Public) ou o Endpoint(Private) na aba Overview.
   	Port	Para acesso via endpoint público, a porta é 443. Para acesso via endpoint interno na mesma região, a porta é 80.
   	HTTP Path	Defina como /cliservice.
   	Username	O nome de usuário. Insira qualquer nome, como root.
   	Password	Insira as informações do token que você criou.
   	HTTP Scheme	Para acesso via endpoint público, defina como https. Para endpoint interno na mesma região, insira http.

6. No topo da página, escolha . Escreva instruções SQL no editor.

   

Conectar ao Spark Thrift Server usando dbt

O dbt (data build tool) permite que analistas e engenheiros de dados escrevam lógica de transformação baseada em SQL e gerenciem implantações com boas práticas de engenharia de software, como versionamento e testes. Para mais informações, consulte a documentação do dbt.

1. Instale o dbt.

   pip install dbt-spark

   Para usar o conector Hive, instale também:

   pip install dbt-spark[HIVE]

2. Crie um projeto dbt.

   dbt init my_spark_project
   cd my_spark_project

3. Configure o perfil do dbt.

   Defina as informações de conexão com o Spark Thrift Server no arquivo ~/.dbt/profiles.yml:

   Conexão via endpoint público

   my_spark_project:
     target: dev
     outputs:
       dev:
         type: spark
         method: thrift
         host: <endpoint>
         port: 443
         user: <username>
         password: <token>
         schema: default
         connect_retries: 5
         connect_timeout: 60
         retry_all: true
         use_ssl: true
         server_side_parameters:
           "hive.exec.dynamic.partition": "true"
           "hive.exec.dynamic.partition.mode": "nonstrict"

   Conexão via endpoint interno na mesma região

   my_spark_project:
     target: dev
     outputs:
       dev:
         type: spark
         method: thrift
         host: <endpoint>
         port: 80
         user: <username>
         password: <token>
         schema: default
         connect_retries: 5
         connect_timeout: 60
         retry_all: true
         use_ssl: false
         server_side_parameters:
           "hive.exec.dynamic.partition": "true"
           "hive.exec.dynamic.partition.mode": "nonstrict"

4. Teste a conexão.

   dbt debug

   Se a conexão for bem-sucedida, você verá uma saída semelhante à seguinte:

   Connection test: [OK connection ok]

5. Crie um modelo dbt.

   Crie um arquivo SQL no diretório models/, como models/example_model.sql:

   {{ config(materialized='table') }} select col1, col2, current_timestamp() as created_at from {{ ref('source_table') }} where col1 is not null

6. Execute o projeto dbt.

   # Run all models
   dbt run
   
   # Run a specific model
   dbt run --models example_model
   
   # Run tests
   dbt test
   
   # Generate documentation
   dbt docs generate
   dbt docs serve

7. Configure as tabelas de origem e os testes.

   Defina as tabelas de origem e os testes no arquivo models/schema.yml:

   version: 2
   
   sources:
     - name: raw_data
       description: "Raw data from source systems"
       tables:
         - name: source_table
           description: "Source table containing raw data"
           columns:
             - name: col1
               description: "Primary identifier"
               tests:
                 - not_null
                 - unique
             - name: col2
               description: "Data column"
   
   models:
     - name: example_model
       description: "Transformed data model"
       columns:
         - name: col1
           description: "Primary identifier"
           tests:
             - not_null
             - unique
         - name: col2
           description: "Processed data column"
         - name: created_at
           description: "Record creation timestamp"

Observações:

• Verifique se a versão do dbt é compatível com o Spark Thrift Server. Use dbt-spark 1.3.0 ou superior.

• Em ambientes de produção, configure informações sensíveis, como tokens, usando variáveis de ambiente em vez de gravá-las diretamente no arquivo de configuração.

• Se ocorrer um timeout de conexão, ajuste os parâmetros connect_timeout e connect_retries.

• Para conjuntos de dados de grande volume, use um modelo de dados incremental para melhorar o desempenho:

{{ config(
    materialized='incremental',
    unique_key='id',
    incremental_strategy='merge'
) }} select * from source_table {% if is_incremental() %} where updated_at > (select max(updated_at) from {{ this }}) {% endif %}

Após concluir essas configurações, você poderá conectar ao Spark Thrift Server usando o dbt e aproveitar seus recursos para transformação e gerenciamento de dados.