Conexión al servidor de consultas Spark mediante el controlador Spark JDBC

Se aplica a: Motor de chispa

Puedes conectarte al servidor de consultas Spark de las siguientes formas y ejecutar consultas para analizar tus datos.

Antes de empezar

  1. Instalar watsonx.data.
  2. Provisión de motor Spark nativo en watsonx.data.
  3. Descarga el cliente JDBC: queryserver-jdbc-SNAPSHOT-standalone.jar desde el enlace de descarga.
  4. Ejecute el Spark Query Server en el motor Spark. Para crear un nuevo servidor de consultas, consulte Crear un servidor de consultas Spark.
  5. Propiedades de conexión - Haga clic en el menú de tres puntos para Query Server, haga clic en Detalles de conexión y copie los siguientes detalles de conexión:
    • Host
    • URI
    • Instancia
    • Nombre de usuario
    • Su clave API IAM de IBM.

Conexión al servidor de consultas Spark mediante DBeaver ( JDBC client)

Para conectarse al servidor de consultas Spark utilizando un cliente JDBC, como DBeaver, configure el controlador watsonx.data en DBeaver.

  1. Abra DBeaver y, en la barra de menús, haga clic en Base de datos > Administrador de controladores.

  2. Buscar Hive. Puede encontrar Apache Hive 4+ driver en Hadoop categoría.

  3. Pulse Copiar.

  4. Cambia el nombre a Spark watsonx.data.

  5. Cambia los siguientes ajustes :

  6. En la pestaña Configuración,

  7. En la pestaña Bibliotecas- Añada el archivo JAR del servidor de consultas Spark JDBC.

  8. Seleccione Base de datos Navigator, haga clic en Nueva conexión y complete los siguientes pasos:

    1. Seleccione el controlador recién creado.
    2. Haga clic en Conectar por y seleccione URL.
    3. Indique la dirección JDBC URL utilizando el siguiente formato : jdbc:hive2://<HOST>:443/default;instance=<INSTANCE>;httpPath=<URI>.
    4. Seleccione Autenticación, proporcione Nombre de usuario como nombre de usuario y su clave de API de IAM como contraseña.
    5. Guarda y conéctate a la conexión haciendo doble clic.

Conexión al servidor de consultas Spark mediante el código Java ( JDBC Client)

Asegúrese de que su CLASSPATH de Java incluye el controlador JDBC descargado. Por ejemplo:

java -cp queryserver-jdbc-SNAPSHOT-standalone.jar App.java

Puede especificar los valores de los parámetros y utilizar el siguiente código Java para conectarse al servidor de consultas Spark. Cuando utilice la API v2, establezca el parámetro <api_version> en v2; para la API v3, establezca v3.

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;

public class App {
    public static void main(String[] args) throws Exception {
        // Set the below configurations from Connection Details of QueryS Server
        // Exclude having https/http/www, just domain
        String host = "example.com";
        String Instance = "CRN/OR/INSTANCE-ID";
        String uri = "/lakehouse/api/<api_version>/spark_engines/.../query_servers/.../connect/cliservice";
        String user = "EMAIL-ID/OR/USER-ID";
        String apikey = "API-KEY";

        String jdbcUrl = String.format("jdbc:hive2://%s/default;instance=%s;httpPath=%s;", host, Instance, uri);

        // Required if your domain requires SSL certificates
        // This is not required for SaaS, hence comment the below line for SaaS
        // Else, we need provide trust-store path which has the SSL certificates for the host
        jdbcUrl += "sslTrustStore=tech_trust.jks;trustStorePassword=Test@123";

        try {
            // Load the Hive JDBC driver
            Class.forName("com.ibm.wxd.spark.jdbc.QueryServerDriver");

            // Connect to Hive
            Connection con = DriverManager.getConnection(jdbcUrl, user, apikey);
            Statement stmt = con.createStatement();

            System.out.println("Connected to watsonx.data Spark Query Server");

            // Sample query
            String sql = "show databases";

            ResultSet rs = stmt.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            int columnCount = rsmd.getColumnCount();

            // The column count starts from 1
            for (int i = 1; i <= columnCount; i++ ) {
                System.out.println(rsmd.getColumnName(i));
            }

            // Print result
            while (rs.next()) {
                System.out.println(rs.getString(1)); // Or loop through columns
            }

            // Clean up
            rs.close();
            stmt.close();
            con.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

Conexión al servidor de consultas Spark mediante Python ( PyHive JDBC Client)

Para conectarse al servidor de consultas Spark utilizando un programa Python, haga lo siguiente:

  1. Asegúrese de que dispone de la versión Python 3.12 o inferior.

  2. Instale pyHive utilizando pip install thrift "PyHive[hive_pure_sasl]==0.7.0".

  3. Guarde el seguimiento en un archivo como connect.py.

    
import ssl
import thrift
import base64
from pyhive import hive

import requests
import thrift.transport
import thrift.transport.THttpClient

import logging
import contextlib
from http.client import HTTPConnection


# Change the following inputs. When using the v2 API, set the <api_version> parameter to `v2`; for the v3 API, set it to `v3`.
class Credentials:
    host = "https://example.ibm.com"
    uri = "/lakehouse/api/<api_version>/spark_engines/.../query_servers/.../connect/cliservice"
    instance_id = "CRN/OR/INSTANCE-ID"
    username = "EMAIL-ID/OR/USER-ID"
    apikey = "API-KEY"


creds = Credentials()


def disable_ssl(ctx):
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    ssl.SSLContext.verify_mode = property(lambda self: ssl.CERT_NONE, lambda self, newval: None)


def get_access_token(apikey):
    try:
        headers = {
            'Content-Type': 'application/x-www-form-urlencoded',
            'Accept': 'application/json',
        }

        data = {
            'grant_type': 'urn:ibm:params:oauth:grant-type:apikey',
            'apikey': apikey,
        }

        response = requests.post('https://iam.cloud.ibm.com/identity/token', headers=headers, data=data)
        return response.json()['access_token']
    except Exception as inst:
        print('Error in getting access token')
        print(inst)
        exit

ctx = ssl.create_default_context()

## If you require to disable SSL, uncomment the below line
# disable_ssl(ctx)

transport = thrift.transport.THttpClient.THttpClient(
    uri_or_host="{host}:{port}{uri}".format(
        host=creds.host, uri= creds.uri, port=443,
    ),
    ssl_context=ctx,
)

headers = {
    "AuthInstanceId": creds.instance_id
}

if creds.instance_id.isdigit():
    # Software installation
    headers["Authorization"] =  "ZenApiKey " + base64.b64encode(f"{creds.username}:{creds.apikey}".encode('utf-8')).decode('utf-8')
else:
    # Cloud installation
    headers["Authorization"] = "Bearer {}".format(get_access_token(creds.apikey))

transport.setCustomHeaders(headers)

cursor = hive.connect(thrift_transport=transport).cursor()
print("Connected to Spark Query Server")

cursor.execute('show databases')
print(cursor.fetchall())

cursor.close()

  4. Ejecutar utilizando python connect.py.