Spark-connector voor Microsoft Fabric Data Warehouse

Met de Spark-connector voor Fabric Data Warehouse kunnen Spark-ontwikkelaars en gegevenswetenschappers toegang krijgen tot en werken met gegevens uit een magazijn en het SQL-analyse-eindpunt van een lakehouse. De connector biedt de volgende mogelijkheden:

  • U kunt met gegevens werken vanuit een magazijn- of SQL-analyse-eindpunt in dezelfde werkruimte of in meerdere werkruimten.
  • Het SQL-analyse-eindpunt van een lakehouse wordt automatisch ontdekt op basis van de context van de werkruimte.
  • De connector heeft een vereenvoudigde Spark-API, abstraheert de onderliggende complexiteit en werkt met slechts één regel code.
  • Terwijl u toegang hebt tot een tabel of weergave, behoudt de connector beveiligingsmodellen die zijn gedefinieerd op sql-engineniveau. Deze modellen omvatten beveiliging op objectniveau (OLS), beveiliging op rijniveau (RLS) en beveiliging op kolomniveau (CLS).
  • De connector is vooraf geïnstalleerd in de Fabric-runtime, waardoor er geen afzonderlijke installatie meer nodig is.

Verificatie

Microsoft Entra-verificatie is een geïntegreerde verificatiemethode. Gebruikers loggen in op de Fabric-werkruimte en hun inloggegevens worden automatisch doorgegeven aan de SQL-engine voor authenticatie en autorisatie. De referenties worden automatisch toegewezen en gebruikers hoeven geen specifieke configuratieopties op te geven.

Notitie

De Spark-connector voor Fabric Data Warehouse ondersteunt alleen interactieve Microsoft Entra-gebruikersverificatie. Verificatie van service-principals wordt niet ondersteund.

Machtigingen

Om verbinding te maken met de SQL-engine hebben gebruikers ten minste leesrechten nodig (vergelijkbaar met de machtiging verbinding maken in SQL Server) op het datawarehouse- of SQL Analytics-eindpunt (itemniveau). Gebruikers hebben ook gedetailleerde machtigingen op objectniveau nodig om gegevens uit specifieke tabellen of weergaven te kunnen lezen. Voor meer informatie, zie Security in Fabric Data Warehouse.

Codesjablonen en voorbeelden

Een methodehandtekening gebruiken

De volgende opdracht toont de synapsesql methodehandtekening voor de leesaanvraag. Het driedelige tableName argument is vereist voor toegang tot tabellen of weergaven vanuit een magazijn en het SQL Analytics-eindpunt van een lakehouse. Werk het argument bij met de volgende namen, op basis van uw scenario:

  • Deel 1: Naam van het magazijn of meerhuis.
  • Deel 2: Naam van het schema.
  • Deel 3: Naam van de tabel of weergave.
synapsesql(tableName:String="<Part 1.Part 2.Part 3>") => org.apache.spark.sql.DataFrame

Naast het rechtstreeks lezen vanuit een tabel of weergave, kunt u met deze connector ook een aangepaste of passthrough-query opgeven, die wordt doorgegeven aan de SQL-engine en het resultaat wordt geretourneerd naar Spark.

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

Hoewel deze connector automatisch het eindpunt voor het opgegeven magazijn/lakehouse detecteert, kunt u dit doen als u dit expliciet wilt opgeven.

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

Gegevens in dezelfde werkruimte lezen

Belangrijk

Voer deze importinstructies uit aan het begin van uw notebook of voordat u de connector gaat gebruiken:

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

De volgende code is een voorbeeld van het lezen van gegevens uit een tabel of weergave in een Spark DataFrame:

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

De volgende code is een voorbeeld van het lezen van gegevens uit een tabel of weergave in een Spark DataFrame met een limiet voor het aantal rijen van 10:

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

De volgende code is een voorbeeld van het lezen van gegevens uit een tabel of weergave in een Spark DataFrame nadat u een filter hebt toegepast:

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

De volgende code is een voorbeeld van het lezen van gegevens uit een tabel of weergave in een Spark DataFrame voor geselecteerde kolommen:

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

Gegevens lezen in werkruimten

Als u gegevens uit een magazijn of een lakehouse in werkruimten wilt openen en lezen, kunt u de werkruimte-id opgeven waar uw magazijn of lakehouse bestaat en vervolgens lakehouse- of magazijnitem-id. De volgende regel bevat een voorbeeld van het lezen van gegevens uit een tabel of weergave in een Spark DataFrame vanuit het magazijn of lakehouse met de opgegeven werkruimte-id en lakehouse/warehouse-id:

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

Notitie

Wanneer u het notebook uitvoert, zoekt de connector standaard naar het opgegeven magazijn of lakehouse in de werkruimte van het lakehouse dat is gekoppeld aan het notebook. Als u wilt verwijzen naar een magazijn of lakehouse vanuit een andere werkruimte, geeft u de werkruimte-id en lakehouse- of magazijnitem-id op zoals hierboven.

Een Lakehouse-tabel maken op basis van gegevens uit een magazijn

Deze coderegels bieden een voorbeeld voor het lezen van gegevens uit een tabel of weergave in een Spark DataFrame en deze gebruiken om een lakehouse-tabel te maken:

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

Een Spark-dataframe-gegevens naar een magazijntabel schrijven

Deze connector maakt gebruik van een schrijfproces in twee fasen naar een Fabric DW-tabel. In eerste instantie worden de Spark-dataframegegevens gestaged in een tijdelijke opslag, gevolgd door het gebruik van de opdracht COPY INTO om de gegevens op te nemen in de Fabric DW-tabel. Deze aanpak zorgt voor schaalbaarheid met een toenemend gegevensvolume.

Ondersteunde dataframe-opslagmodi

De volgende opslagmodi worden ondersteund bij het schrijven van brongegevens van een dataframe naar een doeltabel in het magazijn:

  • ErrorIfExists (standaardopslagmodus): Als de doeltabel bestaat, wordt de schrijfbewerking afgebroken met een uitzonderingsfout die aan de aanroeper wordt geretourneerd. Anders wordt er een nieuwe tabel gemaakt met gegevens.
  • Negeren: Als de doeltabel bestaat, negeert de schrijfbewerking de schrijfaanvraag zonder een fout te retourneren. Anders wordt er een nieuwe tabel gemaakt met gegevens.
  • Overschrijven: Als de doeltabel bestaat, worden bestaande gegevens in de bestemming vervangen door gegevens. Anders wordt er een nieuwe tabel gemaakt met gegevens.
  • Toevoegen: Als de doeltabel bestaat, worden de nieuwe gegevens eraan toegevoegd. Anders wordt er een nieuwe tabel gemaakt met gegevens.

De volgende code toont voorbeelden van het schrijven van gegevens van Spark-dataframes naar een Fabric DW-tabel:

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

Notitie

De connector ondersteunt uitsluitend schrijven naar een Fabric DW-tabel, omdat het SQL-analytische eindpunt van een lakehouse alleen-lezen is.

Leesbewerkingen parallelliseren voor verbeterde prestaties

Deze connector ondersteunt geparallelliseerde leesbewerkingen om de queryprestaties te verbeteren bij het laden van grote tabellen. Net als spark.read.jdbc kunt u parallellisme inschakelen door een partitiekolom en het bijbehorende waardebereik op te geven. Spark splitst de leesbewerking vervolgens op in meerdere partities die gelijktijdig worden verwerkt.

De volgende code laat zien hoe u parallelle leesbewerkingen kunt inschakelen met behulp van een Spark-notebook:

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

Problemen oplossen

Na voltooiing wordt het leesantwoordfragment weergegeven in de uitvoer van de cel. Een fout in de huidige cel onderbreekt ook de volgende celuitvoeringen van het notitieblok. Gedetailleerde foutinformatie is beschikbaar in de Spark-toepassingslogboeken.

Overwegingen voor het gebruik van deze connector

Momenteel is de connector:

  • Ondersteunt het ophalen of lezen van gegevens uit Fabric-magazijnen en SQL-analyse-eindpunten van Lakehouse-items.
  • Ondersteunt het schrijven van data naar een warehouse-tabel door gebruik te maken van verschillende opslagmodi - deze functie is alleen beschikbaar met de nieuwste GA-runtime, Runtime 2.0.
  • Als Private Link is ingeschakeld, worden lees- en schrijfbewerkingen ondersteund op tenantniveau en werkruimteniveau. Schrijfbewerkingen maken gebruik van JDBC-batchinvoegstrategieën. Zie Private Link-ondersteuning voor de Spark-connector voor meer informatie.
  • Fabric DW ondersteunt Time Travel nu, maar deze connector werkt niet voor een query met syntaxis voor tijdreizen.
  • Behoudt de gebruikshandtekening, zoals de handtekening die wordt geleverd met Apache Spark voor Azure Synapse Analytics voor consistentie. Het is echter niet achterwaarts compatibel om verbinding te maken en te werken met een toegewezen SQL-pool in Azure Synapse Analytics.
  • Kolomnamen met speciale tekens worden verwerkt door een escape-teken toe te voegen voordat de query op basis van de driedelige tabel- of weergavenaam wordt ingediend. In een op aangepaste of passthrough-query gebaseerde lezing moeten gebruikers kolomnamen die speciale tekens bevatten escapen.

Gerelateerde inhoud

  • Laatst bijgewerkt om