Connecteur Spark pour l’entrepôt de données Microsoft Fabric

Le connecteur Spark pour l’entrepôt de données Fabric permet aux scientifiques des données et aux développeurs Spark d’accéder et de travailler avec les données d’un entrepôt et de l'endpoint analytique SQL d’un lakehouse. Le connecteur offre les fonctionnalités suivantes :

• Vous pouvez travailler avec des données provenant d'un entrepôt ou d'un point final d'analyse SQL dans le même espace de travail ou dans plusieurs espaces de travail.
• Le point de terminaison d’analytique SQL d’un lakehouse est détecté automatiquement selon le contexte de l’espace de travail.
• Le connecter possède une API Spark simplifiée, extrait la complexité sous-jacente et fonctionne avec une seule ligne de code.
• Pendant que vous accédez à une table ou à une vue, le connecteur respecte les modèles de sécurité définis au niveau du moteur SQL. Ces modèles incluent la sécurité au niveau objet (OLS), la sécurité au niveau ligne (RLS) et la sécurité au niveau colonne (CLS).
• Le connecteur est préinstallé dans le runtime Fabric, ce qui élimine la nécessité d’une installation distincte.

Authentification

L’authentification Microsoft Entra est une approche d’authentification intégrée. Les utilisateurs se connectent à l’espace de travail Fabric, et leurs identifiants sont automatiquement transmis au moteur SQL pour authentification et autorisation. Les informations d’identification sont automatiquement mappées et les utilisateurs ne sont pas obligés de fournir des options de configuration spécifiques.

autorisations

Pour se connecter au moteur SQL, les utilisateurs ont besoin d’au moins une autorisation d’accès en lecture (similaire à l’autorisation CONNECT dans SQL Server) sur l’entrepôt ou sur le point de terminaison d’analytique SQL (au niveau de l’élément). Les utilisateurs ont également besoin des autorisations granulaires au niveau de l’objet pour lire des données à partir de tables ou de vues spécifiques. Pour en savoir plus, consultez Security in Fabric Data Warehouse.

Modèles et exemples de code

Utiliser une signature de méthode

La commande suivante montre la signature de méthode synapsesql pour la demande de lecture. L’argument tableName en trois parties est requis pour accéder aux tables ou aux vues d’un entrepôt et au point de terminaison d’analytique SQL d’un lakehouse. Mettez à jour l’argument avec les noms suivants, en fonction de votre scénario :

• Partie 1 : nom de l’entrepôt ou du lakehouse.
• Partie 2 : nom du schéma.
• Partie 3 : Nom de la table ou de la vue.

synapsesql(tableName:String="<Part 1.Part 2.Part 3>") => org.apache.spark.sql.DataFrame

En plus de lire directement à partir d’une table ou d’une vue, ce connecteur vous permet également de spécifier une requête personnalisée ou directe, qui est passée au moteur SQL et le résultat est retourné à Spark.

spark.read.option(Constants.DatabaseName, "<warehouse/lakeshouse name>").synapsesql("<T-SQL Query>") => org.apache.spark.sql.DataFrame

Bien que ce connecteur découvre automatiquement le point de terminaison de l’entrepôt/lakehouse spécifié, si vous souhaitez le spécifier explicitement, vous pouvez le faire.

//For warehouse
spark.conf.set("spark.datawarehouse.<warehouse name>.sqlendpoint", "<sql endpoint,port>")
//For lakehouse
spark.conf.set("spark.lakehouse.<lakeshouse name>.sqlendpoint", "<sql endpoint,port>")
//Read from table
spark.read.synapsesql("<warehouse/lakeshouse name>.<schema name>.<table or view name>") 

Lire des données dans le même espace de travail

import com.microsoft.spark.fabric
from com.microsoft.spark.fabric.Constants import Constants  

%%spark  
import com.microsoft.spark.fabric.tds.implicits.read.FabricSparkTDSImplicits._
import com.microsoft.spark.fabric.tds.implicits.write.FabricSparkTDSImplicits._
import com.microsoft.spark.fabric.Constants
import org.apache.spark.sql.SaveMode 

Le code suivant est un exemple de lecture de données à partir d’une table ou d’une vue dans un DataFrame Spark :

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>")

Le code suivant est un exemple de lecture de données à partir d’une table ou d’une vue dans un DataFrame Spark avec une limite de nombre de lignes de 10 :

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").limit(10)

Le code suivant est un exemple de lecture de données à partir d’une table ou d’une vue dans un DataFrame Spark après l’application d’un filtre :

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").filter("column name == 'value'")

Le code suivant est un exemple de lecture de données à partir d’une table ou d’une vue dans un DataFrame Spark pour les colonnes sélectionnées uniquement :

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").select("column A", "Column B")

Lire des données entre les espaces de travail

Pour accéder à et lire des données d’un entrepôt ou d’un lakehouse entre des espaces de travail, vous pouvez spécifier l’ID d’espace de travail où se trouve votre entrepôt ou votre lakehouse, puis l’ID de l’élément du lakehouse ou de l’entrepôt. La ligne suivante fournit un exemple de lecture de données d’une table ou d’une vue dans un DataFrame Spark depuis l’entrepôt ou le lakehouse avec l’ID d’espace de travail et l’ID du lakehouse/entrepôt spécifiés :

# For lakehouse
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").synapsesql("<lakehouse name>.<schema name>.<table or view name>")
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").option(Constants.LakehouseId, "<lakehouse item id>").synapsesql("<lakehouse name>.<schema name>.<table or view name>")

# For warehouse
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").synapsesql("<warehouse name>.<schema name>.<table or view name>")
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").option(Constants.DatawarehouseId, "<warehouse item id>").synapsesql("<warehouse name>.<schema name>.<table or view name>")

Créer une table lakehouse basée sur les données d’un entrepôt

Ces lignes de code fournissent un exemple de lecture de données à partir d’une table ou d’une vue dans un DataFrame Spark et l’utilisent pour créer une table lakehouse :

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>")
df.write.format("delta").saveAsTable("<Lakehouse table name>")

Écrire des données de dataframe Spark dans une table d’entrepôt

Ce connecteur utilise un processus d’écriture en deux phases dans une table Fabric DW. Au départ, les données du dataframe Spark sont placées dans un stockage intermédiaire, suivi de la commande COPY INTO pour ingérer les données dans la table Fabric DW. Cette approche garantit l’extensibilité avec l’augmentation du volume de données.

Modes d’enregistrement de DataFrame pris en charge

Les modes d’enregistrement suivants sont pris en charge lors de l’écriture de données sources d’un dataframe dans une table de destination dans l’entrepôt :

• ErrorIfExists (mode d’enregistrement par défaut) : si la table de destination existe, l’écriture est abandonnée avec une exception renvoyée à l'appelant. Sinon, une nouvelle table est créée avec des données.
• Ignorer : si la table de destination existe, la demande d'écriture est ignorée sans retourner d'erreur. Sinon, une nouvelle table est créée avec des données.
• Remplacer : si la table de destination existe, les données existantes dans la destination sont remplacées par des données. Sinon, une nouvelle table est créée avec des données.
• Ajout : si la table de destination existe, les nouvelles données sont ajoutées. Sinon, une nouvelle table est créée avec des données.

Le code suivant montre des exemples d’écriture des données d’un dataframe Spark dans une table Fabric DW :

df.write.synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>") # this uses default mode - errorifexists

df.write.mode("errorifexists").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("ignore").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("append").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("overwrite").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")

Parallélisation des lectures pour améliorer les performances

Ce connecteur prend en charge les lectures parallélisées pour améliorer les performances des requêtes lors du chargement de tables volumineuses. Comme spark.read.jdbc, vous pouvez activer le parallélisme en spécifiant une colonne de partition et sa plage de valeurs. Spark fractionne ensuite l’opération de lecture en plusieurs partitions traitées simultanément.

Le code suivant montre comment activer les lectures parallèles à l’aide d’un notebook Spark :

import com.microsoft.spark.fabric.tds.implicits.read.FabricSparkTDSImplicits._
import com.microsoft.spark.fabric.tds.implicits.write.FabricSparkTDSImplicits._
import com.microsoft.spark.fabric.Constants
import org.apache.spark.sql.SaveMode 

val df = spark.read.option("partitionColumn", <SomeColumn>)

        .option("lowerBound",  <ColumnValuesLowerLimit>)

        .option("upperBound", <ColumnValuesUpperLimit>)

        .option("numPartitions", <NumberOfPartitionsDesired>).synapsesql(<Table>)

Résolution des problèmes

Une fois l’opération terminée, l’extrait de la réponse de lecture s’affiche dans la sortie de la cellule. Un échec dans la cellule active annule les exécutions de la cellule suivante du notebook. Des informations détaillées sur les erreurs sont disponibles dans les journaux de l’application Spark.

Considérations relatives à l’utilisation de ce connecteur

Actuellement, le connecteur :

• Prend en charge la récupération ou la lecture de données dans des entrepôts Fabric et des points de terminaison d’analytique SQL des éléments du lakehouse.
• Prend en charge, l’écriture de données dans une table d’entrepôt en utilisant différents modes de sauvegarde – cette fonctionnalité n’est disponible qu’avec la dernière exécution GA, Runtime 2.0.
• Avec Private Link activé, les opérations de lecture et d’écriture sont prises en charge au niveau du locataire et de l’espace de travail. Les opérations d’écriture utilisent des stratégies d’insertion par lots via JDBC. Pour plus d’informations, consultez la prise en charge de Private Link pour le connecteur Spark.
• Fabric DW prend désormais en charge Time Travel. Toutefois, ce connecteur ne fonctionne pas pour une requête avec une syntaxe de voyage dans le temps.
• Conserve la signature d’utilisation comme celle fournie avec Apache Spark pour Azure Synapse Analytics pour la cohérence. Cependant, il n'est pas rétrocompatible pour se connecter à et travailler avec un pool SQL dédié dans Azure Synapse Analytics.
• Les noms de colonnes avec des caractères spéciaux sont gérés en ajoutant un caractère d’échappement avant l’envoi de la requête, basée sur un nom en 3 parties de la table ou de la vue. Dans une lecture personnalisée ou basée sur une requête directe, les utilisateurs sont tenus d’échapper des noms de colonnes qui contiennent des caractères spéciaux.

Contenu connexe

• Prise en charge de Private Link pour le connecteur Spark
• Runtimes Apache Spark dans Fabric
• Vue d’ensemble de la supervision d'Apache Spark
• Sécurité dans Fabric Data Warehouse
• Autorisations granulaires SQL dans Fabric