E-MapReduce:Manage Spark Thrift Server sessions

Se connecter à Spark Thrift Server avec Python

1. Exécutez la commande suivante pour installer les packages PyHive et Thrift.

   pip install pyhive thrift

2. Rédigez un script Python pour vous connecter à Spark Thrift Server.

   Le script suivant se connecte à Hive et répertorie toutes les bases de données. Choisissez une méthode de connexion adaptée à votre environnement réseau.

   Se connecter via un endpoint public

   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()

   Se connecter via un endpoint interne de la même région

   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()

Se connecter à Spark Thrift Server avec Java

1. Ajoutez les dépendances Maven suivantes à votre fichier 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>
   

   Remarque

   La version intégrée de Hive pour Serverless Spark est la 2.x. Par conséquent, seule la version 2.x de hive-jdbc est prise en charge.

2. Écrivez le code Java nécessaire pour vous connecter à Spark Thrift Server.

   Le code suivant établit une connexion à Spark Thrift Server et interroge la liste des bases de données.

   Se connecter via un endpoint public

   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();
       }
   }

   Se connecter via un endpoint interne de la même région

   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();
       }
   }

Se connecter à Spark Thrift Server avec Spark Beeline

• Si vous utilisez un cluster autogéré, accédez d'abord au répertoire bin de Spark, puis utilisez Beeline pour vous connecter à Spark Thrift Server.

  Se connecter via un endpoint public

  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>"

  Se connecter via un endpoint interne de la même région

  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>"

  Remarque

  Le chemin /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3 indiqué dans le code est un exemple de chemin d'installation de Spark sur un cluster EMR on ECS. Ajustez ce chemin en fonction du chemin d'installation réel de Spark sur votre client. Si vous ne connaissez pas le chemin d'installation de Spark, vous pouvez le trouver en exécutant la commande env | grep SPARK_HOME.

• Si vous utilisez un cluster EMR on ECS, vous pouvez utiliser directement le client Spark Beeline pour vous connecter à Spark Thrift Server.

  Se connecter via un endpoint public

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

  Se connecter via un endpoint interne de la même région

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

Si l'erreur suivante survient lors de la connexion à Serverless Spark Thrift Server avec Hive Beeline, cela est généralement dû à une incompatibilité entre la version de Hive Beeline et Spark Thrift Server. Pour résoudre ce problème, utilisez une version Hive 2.x de Beeline.

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

Configurer Apache Superset pour se connecter à Spark Thrift Server

Apache Superset est une plateforme d'exploration et de visualisation des données qui prend en charge une large gamme de types de graphiques. Pour plus d'informations, consultez la documentation Superset.

1. Installez les dépendances.

   Assurez-vous d'avoir installé la version 0.20.0 du package thrift. Si ce n'est pas le cas, exécutez la commande suivante pour l'installer.

   pip install thrift==0.20.0

2. Démarrez Superset et accédez à l'interface Superset.

   Pour savoir comment démarrer Superset, consultez la documentation Superset.

3. Dans le coin supérieur droit de la page, cliquez sur DATABASE pour accéder à la page Connect a database.

4. Sur la page Connect a database, sélectionnez Apache Spark SQL.

   

5. Saisissez la chaîne de connexion et configurez les paramètres de la source de données.

   Se connecter via un endpoint public

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

   Se connecter via un endpoint interne de même région

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

6. Cliquez sur FINISH pour confirmer la connexion et la validation.

Configurer Hue pour se connecter à Spark Thrift Server

Hue est une interface web open source permettant d'interagir avec l'écosystème Hadoop. Pour plus d'informations, consultez la documentation Hue.

1. Installez les dépendances.

   Assurez-vous d'avoir installé la version 0.20.0 du package thrift. Si ce n'est pas le cas, exécutez la commande suivante pour l'installer.

   pip install thrift==0.20.0

2. Ajoutez la chaîne de connexion Spark SQL au fichier de configuration de Hue.

   Localisez le fichier de configuration de Hue (généralement situé dans /etc/hue/hue.conf), puis ajoutez le contenu suivant au fichier.

   Se connecter via un endpoint public

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

   Se connecter via un endpoint interne de même région

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

3. Redémarrez Hue.

   Après avoir modifié la configuration, vous devez exécuter la commande suivante pour redémarrer le service Hue afin que les modifications prennent effet.

   sudo service hue restart

4. Vérifiez la connexion.

   Une fois le redémarrage effectué avec succès, accédez à l'interface Hue et recherchez l'option Spark SQL. Si la configuration est correcte, vous pouvez vous connecter à Spark Thrift Server et exécuter des requêtes SQL.

   

Se connecter à Spark Thrift Server avec DataGrip

DataGrip est un outil de gestion de base de données permettant d'interroger, de créer et de gérer des bases de données dans des environnements locaux, distants ou cloud. Pour plus d'informations, consultez le site web de DataGrip.

1. Installez DataGrip. Pour plus d'informations, consultez Installer DataGrip.

   La version de DataGrip utilisée dans cet exemple est la 2025.1.2.

2. Ouvrez le client DataGrip. L'interface DataGrip s'affiche.

3. Créez un projet.

   1. Cliquez sur et choisissez Endpoint(Public).

      

   2. Dans la boîte de dialogue New Project, saisissez un nom de projet, par exemple Spark, puis cliquez sur OK.

4. Cliquez sur l'icône dans la barre de menu Database Explorer. Sélectionnez Data Source > Other > Apache Spark.

   

5. Dans la boîte de dialogue Data Sources and Drivers, configurez les paramètres suivants.

   

   Onglet	Paramètre	Description
   General	Name	Un nom de connexion personnalisé. Par exemple, spark_thrift_server.
   	Authentication	Sélectionnez une méthode d'authentification. Dans cette rubrique, l'option No auth est sélectionnée. Dans un environnement de production, sélectionnez User & Password pour garantir que seuls les utilisateurs autorisés peuvent soumettre des jobs SQL, ce qui améliore la sécurité du système.
   	Driver	Cliquez sur Apache Spark, puis sur Go to Driver pour confirmer que la version du pilote est ver. 1.2.2. Remarque Étant donné que la version actuelle du moteur Serverless Spark est la 3.x, vous devez sélectionner la version 1.2.2 du pilote pour garantir la stabilité du système et la compatibilité des fonctionnalités.
   	Endpoint(Private)	L'URL de connexion à Spark Thrift Server. Choisissez une méthode de connexion en fonction de votre environnement réseau. Se connecter via un endpoint public jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token> Se connecter via un endpoint interne de même région jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>
   Options	Run keep-alive query	Facultatif. Activez cette option pour éviter les déconnexions automatiques dues à un délai d'inactivité.

6. Cliquez sur Test Connection pour vérifier la connexion.

   

7. Cliquez sur OK pour terminer la configuration.

8. Vous pouvez désormais gérer Spark Thrift Server à l'aide de DataGrip.

   Une fois DataGrip connecté à Spark Thrift Server, vous pouvez effectuer du développement de données. Pour plus d'informations, consultez la documentation d'aide DataGrip.

   Par exemple, sous la connexion créée, faites un clic droit sur la table cible, choisissez Overview, puis rédigez et exécutez un script SQL dans l'éditeur SQL qui s'ouvre pour afficher les données de la table.

   

Se connecter à Spark Thrift Server avec Redash

Redash est un outil de BI open source permettant d'interroger des bases de données et de visualiser des données via une interface web. Pour plus d'informations, consultez la documentation Redash.

1. Installez Redash. Pour en savoir plus, reportez-vous à la documentation officielle de Redash.

2. Installez les dépendances.

   Assurez-vous d'avoir installé la version 0.20.0 du package thrift. Si ce n'est pas le cas, exécutez la commande suivante pour l'installer.

   pip install thrift==0.20.0

3. Connectez-vous à Redash.

4. Dans le volet de navigation de gauche, cliquez sur Settings, puis sur l'onglet Data Sources, sélectionnez +New Data Source.

5. Dans la boîte de dialogue qui s'affiche, configurez les paramètres, puis cliquez sur Create.

   

   Parameter		Description
   Type Selection		Le type de source de données. Dans la zone de recherche, trouvez et sélectionnez Token Management.
   Configuration	Name	Le nom de la source de données. Vous pouvez personnaliser ce nom.
   	Host	L'endpoint de Spark Thrift Server. Vous pouvez obtenir l'Token Management ou l'New dans l'onglet URL.
   	Port	Si vous accédez via l'endpoint public, le numéro de port est 443. Si vous accédez via l'endpoint interne de même région, le numéro de port est 80.
   	HTTP Path	Définissez cette valeur sur /cliservice.
   	Username	Le nom d'utilisateur. Vous pouvez saisir n'importe quel nom, par exemple root.
   	Password	Saisissez les informations de token que vous avez créées.
   	HTTP Scheme	Si vous accédez via l'endpoint public, définissez cette valeur sur https. Pour un endpoint interne de même région, saisissez http.

6. En haut de la page, choisissez . Vous pouvez rédiger des instructions SQL dans l'éditeur.

   

Se connecter à Spark Thrift Server avec dbt

dbt (data build tool) permet aux analystes de données et aux ingénieurs d'écrire une logique de transformation basée sur SQL et de gérer les déploiements selon les bonnes pratiques du génie logiciel, telles que le versioning et les tests. Pour plus d'informations, consultez la documentation dbt.

1. Installez dbt.

   pip install dbt-spark

   Pour utiliser le connecteur Hive, vous pouvez également installer :

   pip install dbt-spark[HIVE]

2. Créez un projet dbt.

   dbt init my_spark_project
   cd my_spark_project

3. Configurez le profil dbt.

   Configurez les informations de connexion à Spark Thrift Server dans le fichier ~/.dbt/profiles.yml :

   Connexion via un endpoint public

   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"

   Connexion via un endpoint interne de même région

   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. Testez la connexion.

   dbt debug

   Si la connexion aboutit, vous obtiendrez une sortie similaire à la suivante :

   Connection test: [OK connection ok]

5. Créez un modèle dbt.

   Créez un fichier SQL dans le répertoire models/, par exemple 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. Exécutez le projet 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. Configurez les tables sources et les tests.

   Définissez les tables sources et les tests dans le fichier 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"

Remarques :

• Assurez-vous que votre version de dbt est compatible avec Spark Thrift Server. Utilisez dbt-spark 1.3.0 ou une version ultérieure.

• Dans un environnement de production, configurez les informations sensibles, telles que les tokens, à l'aide de variables d'environnement plutôt que de les écrire directement dans le fichier de configuration.

• En cas de délai d'expiration de la connexion, ajustez les paramètres connect_timeout et connect_retries.

• Pour les grands ensembles de données, utilisez un modèle de données incrémentiel afin d'améliorer les performances :

{{ 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 %}

Une fois ces configurations effectuées, vous pouvez vous connecter à Spark Thrift Server via dbt et exploiter ses fonctionnalités avancées pour la transformation et la gestion des données.