Connettore Apache Spark: SQL Server e Azure SQL

Il connettore Apache Spark per SQL Server e Azure SQL è un connettore ad alte prestazioni che consente di usare i dati transazionali in analisi dei Big Data e di salvare i risultati per la creazione di query o report ad hoc. Il connettore consente di usare qualsiasi database SQL, in locale o nel cloud, come origine dati di input o sink di dati di output per i processi Spark.

Questa libreria contiene il codice sorgente per il connettore Apache Spark per SQL Server e Azure SQL.

Apache Spark è un motore di analisi unificato per l'elaborazione di dati su larga scala.

Sono disponibili due versioni del connettore tramite Maven, una compatibile con 2.4.x e una compatibile con 3.0.x. Entrambe le versioni sono disponibili qui e possono essere importate usando le coordinate seguenti:

È anche possibile creare il connettore dall'origine o scaricare il file con estensione jar dalla sezione Release in GitHub. Per informazioni aggiornate sul connettore, vedere il repository GitHub del connettore SQL Spark.

Funzionalità supportate

• Supporto di tutte le associazioni Spark (Scala, Python, R)
• Supporto di autenticazione di base e keytab Active Directory (AD)
• Supporto della scrittura di dataframe riordinato
• Supporto della scrittura in Istanza singola e Pool di dati di SQL Server in cluster Big Data di SQL Server
• Supporto di Reliable Connector per Istanza singola di SQL Server

Opzioni supportate

Il connettore Apache Spark per SQL Server e Azure SQL supporta le opzioni definite qui: SQL DataSource JDBC

Sono supportate anche le opzioni seguenti:

Altre opzioni di copia bulk possono essere impostate come opzioni in dataframe e verranno passate alle API bulkcopy durante la scrittura

Confronto delle prestazioni

Il connettore Apache Spark per SQL Server e Azure SQL è 15 volte più veloce del connettore JDBC generico per la scrittura in SQL Server. Le prestazioni variano a seconda del tipo, del volume di dati, delle opzioni usate e possono far registrare variazioni da un'esecuzione all'altra. I risultati di prestazioni seguenti corrispondono al tempo impiegato per sovrascrivere una tabella SQL con 143,9 milioni di righe in un dataframe Spark. Il dataframe Spark viene costruito leggendo la tabella HDFS store_sales generata con il benchmark TPCDS Spark. Il tempo per la lettura da store_sales a dataframe non è incluso. I risultati corrispondono a una media calcolata su tre esecuzioni.

Config

• Configurazione Spark: num_executors = 20, executor_memory = '1664 m', executor_cores = 2
• Configurazione generatore dati: scale_factor=50, partitioned_tables=true
• File di dati store_sales con 143.997.590 righe

Ambiente

• Cluster Big Data di SQL Server CU5
• master + 6 nodi
• Ogni nodo con server gen 5, 512 GB di RAM, 4 TB di NVM per nodo, NIC 10 GB

Problemi comuni

java.lang.NoClassDefFoundError: com/microsoft/aad/adal4j/AuthenticationException

Questo problema si verifica quando si usa una versione precedente del driver MSSQL (ora incluso in questo connettore) nell'ambiente Hadoop in uso. Se prima di questo momento veniva usato il connettore Azure SQL precedente e sono stati installati manualmente driver nel cluster per la compatibilità con l’autenticazione Microsoft Entra, sarà necessario rimuovere tali driver.

Passaggi per risolvere il problema:

1. Se si usa un ambiente Hadoop generico, controllare e rimuovere il file MSSQL con estensione jar: rm $HADOOP_HOME/share/hadoop/yarn/lib/mssql-jdbc-6.2.1.jre7.jar. Se si usa Databricks, aggiungere uno script di inizializzazione globale o cluster per rimuovere le versioni precedenti del driver MSSQL dalla cartella /databricks/jars o aggiungere la riga seguente a uno script esistente: rm /databricks/jars/*mssql*

2. Aggiungere i pacchetti adal4j e mssql. Ad esempio è possibile usare Maven, ma qualsiasi modalità dovrebbe funzionare.

   Attenzione

   Non installare il connettore Spark SQL in questo modo.

3. Aggiungere la classe driver alla configurazione della connessione. Ad esempio:

   connectionProperties = {
     `Driver`: `com.microsoft.sqlserver.jdbc.SQLServerDriver`
   }`
   

Per altre informazioni e spiegazioni, vedere la pagina relativa alla risoluzione di https://github.com/microsoft/sql-spark-connector/issues/26.

Operazioni preliminari

Il connettore Apache Spark per SQL Server e Azure SQL si basa sull'API Spark DataSourceV1 e sull'API SQL Server Bulk e usa la stessa interfaccia del connettore Spark SQL JDBC incorporato. Questa integrazione consente di integrare facilmente il connettore ed eseguire la migrazione dei processi Spark esistenti semplicemente aggiornando il parametro format con com.microsoft.sqlserver.jdbc.spark.

Per includere il connettore nei progetti, scaricare questo repository e compilare il file con estensione jar usando SBT.

Scrivere in una nuova tabella SQL

Quando si usa la modalità overwrite, se non si usa l'opzione truncate quando si ricrea la tabella, gli indici vanno perduti. Una tabella columnstore diventa ora un heap. Se si vuole mantenere l'indicizzazione esistente, specificare anche l'opzione truncate con valore true. Ad esempio, .option("truncate","true").

server_name = "jdbc:sqlserver://{SERVER_ADDR}"
database_name = "database_name"
url = server_name + ";" + "databaseName=" + database_name + ";"

table_name = "table_name"
username = "username"
password = "password123!#" # Please specify password here

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("overwrite") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

Accodare alla tabella SQL

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("append") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

Specificare il livello di isolamento

Per impostazione predefinita, questo connettore usa il livello di isolamento READ_COMMITTED quando si esegue l'inserimento bulk nel database. Se si vuole eseguire l'override del livello di isolamento, usare l'opzione mssqlIsolationLevel come illustrato di seguito.

    .option("mssqlIsolationLevel", "READ_UNCOMMITTED") \

Leggere dalla tabella SQL

jdbcDF = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("user", username) \
        .option("password", password).load()

Autenticazione Microsoft Entra

Esempio di Python con entità servizio

context = adal.AuthenticationContext(authority)
token = context.acquire_token_with_client_credentials(resource_app_id_url, service_principal_id, service_principal_secret)
access_token = token["accessToken"]

jdbc_db = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("accessToken", access_token) \
        .option("encrypt", "true") \
        .option("hostNameInCertificate", "*.database.windows.net") \
        .load()

Esempio di Python con password di Active Directory

jdbc_df = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("authentication", "ActiveDirectoryPassword") \
        .option("user", user_name) \
        .option("password", password) \
        .option("encrypt", "true") \
        .option("hostNameInCertificate", "*.database.windows.net") \
        .load()

Per eseguire l'autenticazione con Active Directory, è necessario installare una dipendenza obbligatoria.

Il formato di user quando si usa ActiveDirectoryPassword deve essere il formato UPN, ad esempio username@domainname.com.

Per Scala è necessario installare l'artefatto _com.microsoft.aad.adal4j_.

Per Python è necessario installare la libreria _adal_. Questa operazione è disponibile tramite PIP.

Per esempi, vedere Notebook di esempio.

Supporto tecnico

Il connettore Apache Spark per Azure SQL e SQL Server è un progetto open source. Questo connettore non dispone di supporto tecnico Microsoft. Per problemi o domande sul connettore, creare una voce nel repository del progetto. La community del connettore è attiva ed elabora le domande inoltrate.

Passaggi successivi

Visitare il repository GitHub del connettore Spark SQL.

Per informazioni sui livelli di isolamento, vedere SET TRANSACTION ISOLATION LEVEL (Transact-SQL).