Connessione alle origini dati

Managed Service for Apache Spark si integra con database, data warehouse e servizi di streaming per elaborare i dati provenienti da più sistemi. Questo documento mostra come connettere i cluster Managed Service for Apache Spark alle origini dati all'interno e all'esterno di Google Cloud.

Prima di iniziare

  1. Accedi al tuo Google Cloud account. Se non hai mai utilizzato Google Cloud, crea un account per valutare il rendimento dei nostri prodotti in scenari reali. I nuovi clienti ricevono anche 300 $di crediti senza costi per eseguire, testare ed eseguire il deployment dei carichi di lavoro.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that you have the permissions required to complete this guide.

  4. Verify that billing is enabled for your Google Cloud project.

  5. Enable the Managed Service for Apache Spark, BigQuery, and Cloud Storage APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

  6. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  7. Verify that you have the permissions required to complete this guide.

  8. Verify that billing is enabled for your Google Cloud project.

  9. Enable the Managed Service for Apache Spark, BigQuery, and Cloud Storage APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

    10.
  10. Crea un cluster Managed Service for Apache Spark.
  11. Crea un bucket Cloud Storage.
  12. Assicurati di disporre delle autorizzazioni necessarie per accedere alle origini dati.

Ruoli obbligatori

Per eseguire gli esempi in questa pagina sono necessari alcuni ruoli Identity and Access Management (IAM). A seconda delle norme dell'organizzazione, questi ruoli potrebbero essere già stati concessi. Per verificare le concessioni dei ruoli, consulta la sezione Devi concedere i ruoli?.

Per ulteriori informazioni sulla concessione dei ruoli, consulta Gestire l'accesso a progetti, cartelle e organizzazioni.

Ruoli utente

Per ottenere le autorizzazioni necessarie per creare un cluster Managed Service for Apache Spark, chiedi all'amministratore di concederti i seguenti ruoli IAM:

Ruolo service account

Per assicurarti che il account di servizio predefinito di Compute Engine disponga delle autorizzazioni necessarie per creare un cluster Managed Service for Apache Spark, chiedi all'amministratore di concedere il ruolo IAM Dataproc Worker (roles/dataproc.worker) al account di servizio predefinito di Compute Engine sul progetto.

Connessione ai Google Cloud servizi

Questa sezione mostra come connettere Managed Service for Apache Spark a Google Cloud e alle origini dati esterne.

Connessione a Cloud Storage

Managed Service for Apache Spark utilizza Cloud Storage come file system distribuito predefinito. Per leggere e scrivere dati da e nei bucket Cloud Storage, utilizza il prefisso URI gs://. Non è richiesta alcuna configurazione aggiuntiva.

L'esempio seguente mostra come leggere un file CSV da un bucket Cloud Storage e quindi scrivere un file Parquet nel bucket.

# Read a CSV file from a Cloud Storage bucket into a Spark DataFrame
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("gs://BUCKET_NAME/path/to/data.csv")

df.show()

# Write a Spark DataFrame to a Cloud Storage bucket
df.write.format("parquet") \
    .mode("overwrite") \
    .save("gs://BUCKET_NAME/path/to/output.parquet")

Connessione a BigQuery

Il connettore BigQuery è preinstallato nelle versioni dell'immagine Managed Service for Apache Spark 2.1 e successive. Il connettore consente il trasferimento di dati su larga scala tra Managed Service for Apache Spark e BigQuery.

L'esempio seguente mostra come leggere una tabella BigQuery in un DataFrame Spark e quindi eseguire una query.

# Set up the BigQuery connector.
spark.conf.set("spark.sql.sources.provider", "bigquery")
spark.conf.set("viewsEnabled", "true")
spark.conf.set("materializationDataset", "YOUR_BQ_DATASET")

# Read a BigQuery table into a Spark DataFrame,
df = spark.read.format("bigquery") \
    .option("table", "bigquery-public-data.samples.shakespeare") \
    .load()

df.createOrReplaceTempView("shakespeare")

# Perform a query on the data,
word_counts = spark.sql("SELECT word, SUM(word_count) as count FROM shakespeare GROUP BY word ORDER BY count DESC")
word_counts.show()

Connessione a database esterni tramite JDBC

Puoi connettere Managed Service for Apache Spark a un database che fornisce un driver Java Database Connectivity (JDBC), come PostgreSQL o MySQL, utilizzando l'origine dati JDBC integrata di Spark.

Rendere disponibile il driver JDBC

Per connetterti a un database con JDBC, rendi disponibile il file JAR del driver nel classpath di Spark.

Per job

Archivia il file JAR del driver in un bucket Cloud Storage e fai riferimento a questo file con il flag --jars quando invii un job. Spark distribuisce il file JAR ai nodi necessari per il job. Questo è l'approccio consigliato per le dipendenze specifiche del job.

A livello di cluster

Se tutti i job di un cluster richiedono un driver specifico, configura la spark:spark.jars proprietà del cluster quando crei il cluster per fare riferimento al file JAR del driver in Cloud Storage. In questo modo, la dipendenza viene inserita nei classpath del driver e dell'executor di Spark, mantenendo al contempo l'isolamento del classpath per l'agente Managed Service for Apache Spark.

Crea il cluster con la proprietà del cluster spark:spark.jars:

gcloud dataproc clusters create my-jdbc-cluster \
    --region=REGION \
    --image-version=2.2-debian11 \
    --properties="spark:spark.jars=gs://YOUR_BUCKET_NAME/drivers/postgresql-42.7.3.jar"

Lettura da un database PostgreSQL

Questo esempio utilizza il metodo per job per connettersi a un database PostgreSQL.

  1. Scarica il file JAR del driver JDBC PostgreSQL e caricalo in un bucket Cloud Storage.

  2. Invia il job PySpark e fai riferimento al percorso Cloud Storage del driver con il flag --jars.

    gcloud dataproc jobs submit pyspark my_job.py \
    --cluster=MY_CLUSTER \
    --region=REGION \
    --jars=gs://YOUR_BUCKET_NAME/drivers/postgresql-42.7.3.jar

  3. Utilizza il seguente codice nel file job.py per leggere e scrivere nel database.

    from pyspark.sql import SparkSession

def main():
    # WARNING: Do not hardcode credentials. Use a service like Secret Manager
    # to handle sensitive credentials.
    db_properties = {
        "user": "USERNAME",
        "password": "PASSWORD",
        "driver": "org.postgresql.Driver"
    }

    jdbc_url = "jdbc:postgresql://DB_HOST:DB_PORT/DATABASE"

    spark = SparkSession.builder.appName("Postgres JDBC Example").getOrCreate()

    # Read data from a PostgreSQL table
    df = spark.read.jdbc(url=jdbc_url, table="public.my_table", properties=db_properties)

    df.printSchema()
    df.show()

    # Write data to a new PostgreSQL table
    (df.write
        .jdbc(url=jdbc_url,
              table="public.new_table",
              mode="overwrite",
              properties=db_properties))

if __name__ == "__main__":
    main()

Connessione a database esterni tramite ODBC

Per connetterti alle origini dati in cui non è disponibile un driver JDBC, utilizza un driver Open Database Connectivity (ODBC). Utilizza un'azione di inizializzazione per installare il driver e le relative dipendenze su ogni nodo quando crei il cluster.

Questo esempio mostra come connettersi a un'istanza di Microsoft SQL Server.

  1. Crea lo script dell'azione di inizializzazione.

    1. Crea uno script che installa il driver ODBC di Microsoft SQL Server per Debian.
        #!/bin/bash
    # Initialization action for installing MS SQL ODBC driver on the cluster.

    set -e -x

    # Install dependencies for the driver.
    apt-get update
    apt-get install -y --no-install-recommends curl gnupg unixodbc-dev

    # Add Microsoft's official repository.
    curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add -
    curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list

    # Install the driver.
    apt-get update
    ACCEPT_EULA=Y apt-get install -y msodbcsql18

    # Install the pyodbc library for Python.
    pip install pyodbc
    ```

1.  Upload the script to a Cloud Storage bucket.

```shell
    gsutil cp install-odbc-mssql.sh gs://YOUR_BUCKET_NAME/init-actions/
    ```

  2. Crea un cluster Managed Service for Apache Spark che esegue l'azione di inizializzazione.

    Quando crei il cluster, fai riferimento allo script dell'azione di inizializzazione in Cloud Storage.

    gcloud dataproc clusters create my-odbc-cluster \
    --region=REGION \
    --image-version=2.2-debian11 \
    --initialization-actions=gs://BUCKET_NAME/init-actions/install-odbc-mssql.sh

  3. Connessione ed esecuzione di query tramite PySpark

    Una volta che il cluster è in esecuzione, il driver ODBC e la libreria pyodbc sono disponibili. Il seguente codice utilizza pyodbc sul nodo del driver per recuperare i dati in un DataFrame Pandas e poi li converte in un DataFrame Spark per l'elaborazione distribuita.

    import pyodbc
import pandas as pd
from pyspark.sql import SparkSession

def get_sql_data(connection_string, query):
    """
    Connects to the database using pyodbc, executes a query,
    and returns the result as a Pandas DataFrame.
    """
    cnxn = pyodbc.connect(connection_string)
    pdf = pd.read_sql(query, cnxn)
    cnxn.close()
    return pdf

def main():
    spark = SparkSession.builder.appName("ODBC Example").getOrCreate()

    # WARNING: Do not hardcode credentials. Use Secret Manager.
    server = 'SERVER.database.windows.net'
    database = 'DATABASE'
    username = 'USERNAME'
    password = 'PASSWORD'
    driver = '{ODBC Driver 18 for SQL Server}' # The driver name must match the installed driver.

    # Create the connection string.
    connection_string = f'DRIVER={driver};SERVER=tcp:{server},1433;DATABASE={database};UID={username};PWD={password};Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;'

    sql_query = "SELECT * FROM Sales.Customer"

    # Fetch data as a Pandas DataFrame.
    pandas_df = get_sql_data(connection_string, sql_query)

    # Convert the Pandas DataFrame to a Spark DataFrame for distributed processing.
    spark_df = spark.createDataFrame(pandas_df)

    print("Successfully read data from SQL Server:")
    spark_df.printSchema()
    spark_df.show(5)

    # You can now perform distributed operations on the Spark DataFrame.
    print(f"Total number of customers: {spark_df.count()}")

if __name__ == "__main__":
    main()

Best practice

Questa sezione descrive le best practice per la gestione delle credenziali e delle dipendenze.

Gestione delle credenziali

Non codificare le password o altri secret nel codice. Utilizza Secret Manager per archiviare le credenziali in modo sicuro e accedervi dai job Managed Service for Apache Spark.

Gestione delle dipendenze

Utilizza le azioni di inizializzazione per installare driver e librerie quando crei un cluster. Per gli ambienti complessi, valuta la possibilità di creare un'immagine Managed Service for Apache Spark personalizzata con tutte le dipendenze preinstallate. Un'immagine personalizzata riduce il tempo di avvio del cluster e garantisce la coerenza.

Copia gli script delle azioni di inizializzazione da origini pubbliche nel tuo bucket Cloud Storage con controllo delle versioni prima di utilizzarli in produzione. Non fare riferimento direttamente agli script pubblici, perché possono cambiare senza preavviso.

Passaggi successivi