Σύνδεση Spark για βάσεις δεδομένων SQL

Η σύνδεση Spark για βάσεις δεδομένων SQL είναι μια βιβλιοθήκη υψηλών επιδόσεων που σας επιτρέπει να διαβάζετε και να γράφετε σε βάσεις δεδομένων SQL Server, Azure SQL και βάσεις δεδομένων SQL στο Fabric. Η σύνδεση προσφέρει τις εξής δυνατότητες:

  • Χρησιμοποιήστε το Spark για να εκτελέσετε μεγάλες λειτουργίες εγγραφής και ανάγνωσης σε βάση δεδομένων Azure SQL Database, Azure SQL Managed Instance, SQL Server στο Azure VM και βάσεις δεδομένων SQL στο Fabric.
  • Όταν χρησιμοποιείτε έναν πίνακα ή μια προβολή, η σύνδεση υποστηρίζει μοντέλα ασφαλείας που έχουν οριστεί σε επίπεδο μηχανισμού SQL. Αυτά τα μοντέλα περιλαμβάνουν ασφάλεια σε επίπεδο αντικειμένου (OLS), ασφάλεια σε επίπεδο γραμμών (RLS) και ασφάλεια σε επίπεδο στηλών (CLS).

Η σύνδεση είναι προεγκατεστημένη στο χρόνο εκτέλεσης Fabric, επομένως δεν χρειάζεται να την εγκαταστήσετε ξεχωριστά.

Έλεγχος ταυτότητας

Ο έλεγχος ταυτότητας Microsoft Entra είναι ενσωματωμένος στο Fabric.

  • Όταν εισέρχεστε στον χώρο εργασίας Fabric, τα διαπιστευτήριά σας μεταβιβάζονται αυτόματα στον μηχανισμό SQL για έλεγχο ταυτότητας και εξουσιοδότηση.
  • Απαιτεί την ενεργοποίηση και τη ρύθμιση παραμέτρων του Microsoft Entra ID στον μηχανισμό βάσης δεδομένων SQL.
  • Δεν απαιτείται επιπλέον ρύθμιση παραμέτρων στον κώδικα Spark εάν έχει ρυθμιστεί το Microsoft Entra ID. Τα διαπιστευτήρια αντιστοιχίζονται αυτόματα.

Μπορείτε επίσης να χρησιμοποιήσετε τη μέθοδο ελέγχου ταυτότητας SQL (καθορίζοντας ένα όνομα χρήστη και έναν κωδικό πρόσβασης SQL) ή μια κύρια υπηρεσία (παρέχοντας ένα διακριτικό πρόσβασης Azure για έλεγχο ταυτότητας βάσει εφαρμογής).

Δικαιώματα

Για να χρησιμοποιήσετε τη σύνδεση Spark, η ταυτότητά σας, είτε πρόκειται για χρήστη είτε για εφαρμογή, πρέπει να έχει τα απαραίτητα δικαιώματα βάσης δεδομένων για τον μηχανισμό SQL προορισμού. Αυτά τα δικαιώματα απαιτούνται για ανάγνωση ή εγγραφή σε πίνακες και προβολές.

Για τη βάση δεδομένων SQL Azure, τη διαχειριζόμενη παρουσία SQL Azure και τον SQL Server σε εικονική μηχανή Azure:

  • Η ταυτότητα που εκτελεί τη λειτουργία συνήθως χρειάζεται db_datawriter και db_datareader δικαιώματα, και προαιρετικά db_owner για πλήρη έλεγχο.

Για μια βάση δεδομένων SQL στο Fabric:

  • Η ταυτότητα συνήθως χρειάζεται db_datawriter και db_datareader δικαιώματα, και προαιρετικά db_owner.
  • Η ταυτότητα χρειάζεται επίσης τουλάχιστον δικαίωμα ανάγνωσης στη βάση δεδομένων SQL στο Fabric σε επίπεδο στοιχείου.

Σημείωμα

Εάν χρησιμοποιείτε μια κύρια υπηρεσία, μπορεί να εκτελεστεί ως εφαρμογή (χωρίς περιβάλλον χρήστη) ή ως χρήστης, εάν είναι ενεργοποιημένη η απομίμηση χρήστη. Η κύρια υπηρεσία πρέπει να έχει τα απαιτούμενα δικαιώματα βάσης δεδομένων για τις λειτουργίες που θέλετε να εκτελέσετε.

Παραδείγματα χρήσης και κώδικα

Σε αυτήν την ενότητα, παρέχουμε παραδείγματα κώδικα για να δείξουμε πώς να χρησιμοποιείτε αποτελεσματικά τη σύνδεση Spark για βάσεις δεδομένων SQL. Αυτά τα παραδείγματα καλύπτουν διάφορα σενάρια, συμπεριλαμβανομένης της ανάγνωσης και εγγραφής σε πίνακες SQL και της διαμόρφωσης των επιλογών σύνδεσης.

Σημείωμα

Πριν από μια μαζική εγγραφή, όλα τα εισερχόμενα δεδομένα Spark πρέπει να ταιριάζουν στους τύπους δεδομένων SQL προορισμού. Όταν αντικαθιστάτε ή δημιουργείτε έναν πίνακα, η σύνδεση αντιστοιχίζει το Spark TimestampType και TimestampNTZType τις τιμές σε SQL datetime2 αντί για datetime. Οι τύποι χρονικής σήμανσης Spark υποστηρίζουν έως και έξι ψηφία ακρίβειας κλασματικού δευτερολέπτου, αλλά η SQL datetime υποστηρίζει τρία, γεγονός που μπορεί να προκαλέσει αναντιστοιχία.

Υποστηριζόμενες επιλογές

Η ελάχιστη απαιτούμενη επιλογή είναι url as "jdbc:sqlserver://<server>:<port>;database=<database>;" or set spark.mssql.connector.default.url.

  • Όταν το url παρέχεται:

    • Χρησιμοποιείτε url πάντα ως πρώτη προτίμηση.
    • Εάν spark.mssql.connector.default.url δεν έχει οριστεί, η σύνδεση θα το ρυθμίσει και θα το χρησιμοποιήσει ξανά για μελλοντική χρήση.

  • Όταν το url δεν παρέχεται:

    • Εάν spark.mssql.connector.default.url έχει οριστεί, η σύνδεση χρησιμοποιεί την τιμή από τη ρύθμιση παραμέτρων σπινθήρα.
    • Εάν spark.mssql.connector.default.url δεν έχει οριστεί, εμφανίζεται ένα σφάλμα επειδή οι απαιτούμενες λεπτομέρειες δεν είναι διαθέσιμες.

Αυτή η σύνδεση υποστηρίζει τις επιλογές που ορίζονται εδώ: Επιλογές JDBC SQL DataSource

Η υποδοχή υποστηρίζει επίσης τις ακόλουθες επιλογές:

Επιλογή Προεπιλεγμένη τιμή Περιγραφή
reliabilityLevel "BEST_EFFORT" Ελέγχει την αξιοπιστία των λειτουργιών ένθεσης. Πιθανές τιμές: BEST_EFFORT (προεπιλογή, ταχύτερη, μπορεί να οδηγήσει σε διπλότυπες γραμμές σε περίπτωση επανεκκίνησης ενός εκτελεστή), NO_DUPLICATES (πιο αργή, εξασφαλίζει ότι δεν θα εισαχθούν διπλότυπες γραμμές ακόμα και αν γίνει επανεκκίνηση ενός εκτελεστή). Επιλέξτε με βάση την ανοχή σας για διπλότυπα και τις ανάγκες απόδοσης.
isolationLevel "READ_COMMITTED" Ορίζει το επίπεδο απομόνωσης συναλλαγών για λειτουργίες SQL. Πιθανές τιμές: READ_COMMITTED (προεπιλογή, αποτρέπει την ανάγνωση μη δεσμευμένων δεδομένων), READ_UNCOMMITTED, REPEATABLE_READ, SNAPSHOT, SERIALIZABLE. Τα υψηλότερα επίπεδα απομόνωσης μπορούν να μειώσουν τον συγχρονισμό αλλά να βελτιώσουν τη συνέπεια των δεδομένων.
tableLock "ψευδής" Ελέγχει εάν η υπόδειξη κλειδώματος σε επίπεδο πίνακα TABLOCK του SQL Server χρησιμοποιείται κατά τη διάρκεια των λειτουργιών εισαγωγής. Πιθανές τιμές: true (ενεργοποιεί το TABLOCK, το οποίο μπορεί να βελτιώσει την απόδοση μαζικής εγγραφής), false (προεπιλογή, δεν χρησιμοποιεί TABLOCK). Η ρύθμιση σε μπορεί να true αυξήσει την ταχύτητα μετάδοσης για μεγάλα ένθετα, αλλά μπορεί να μειώσει τον ταυτοχρονισμό για άλλες λειτουργίες στον πίνακα.
schemaCheckEnabled "true" Ελέγχει εάν επιβάλλεται αυστηρή επικύρωση σχήματος μεταξύ του Spark DataFrame και του πίνακα SQL. Πιθανές τιμές: true (προεπιλογή, επιβάλλει αυστηρή αντιστοίχιση σχήματος), false (επιτρέπει μεγαλύτερη ευελιξία και ενδέχεται να παραλείψει ορισμένους ελέγχους σχήματος). Η ρύθμιση σε μπορεί να false βοηθήσει με τις αναντιστοιχίες σχήματος, αλλά μπορεί να οδηγήσει σε απροσδόκητα αποτελέσματα, εάν οι δομές διαφέρουν σημαντικά.

Άλλες επιλογές μαζικού API μπορούν να οριστούν ως επιλογές στο και μεταβιβάζονται σε API μαζικής αντιγραφής κατά την DataFrame εγγραφή.

Παράδειγμα γραφής και ανάγνωσης

Ο παρακάτω κώδικας χρησιμοποιεί αυτόματο έλεγχο ταυτότητας Microsoft Entra ID για την επίδειξη αυτών των λειτουργιών:

  • Γράψτε ένα DataFrame: df.write.option("...", "...").mssql("<schema>.<table>").
  • Διαβάστε έναν πίνακα: spark.read.option("...", "...").mssql("<schema>.<table>").
  • Εκτελέστε ένα προσαρμοσμένο ερώτημα: spark.read.option("...", "...").option("query", "<your-custom-query>").mssql().

Φιλοδώρημα

Τα δεδομένα δημιουργούνται ενσωματωμένα για σκοπούς επίδειξης. Σε ένα σενάριο παραγωγής, συνήθως θα διαβάζατε δεδομένα από μια υπάρχουσα πηγή ή θα δημιουργούσατε ένα πιο περίπλοκο DataFrame.

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)
df.write.mode("overwrite").option("url", url).mssql("dbo.publicExample")
spark.read.option("url", url).mssql("dbo.publicExample").show()
spark.read.option("url", url).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

url = "jdbc:sqlserver://<server>:<port>;database=<database2>;" # different database
df.write.mode("overwrite").option("url", url).mssql("dbo.tableInDatabase2") # default url is updated
spark.read.mssql("dbo.tableInDatabase2").show() # no url option specified and will use database2

Μπορείτε επίσης να επιλέξετε στήλες, να εφαρμόσετε φίλτρα και να χρησιμοποιήσετε άλλες επιλογές όταν διαβάζετε δεδομένα από το μηχανισμό βάσης δεδομένων SQL.

Παραδείγματα ελέγχου ταυτότητας

Τα παρακάτω παραδείγματα δείχνουν πώς μπορείτε να χρησιμοποιήσετε μεθόδους ελέγχου ταυτότητας εκτός του Microsoft Entra ID, όπως η κύρια υπηρεσία (διακριτικό πρόσβασης) και ο έλεγχος ταυτότητας SQL.

Σημείωμα

Όπως αναφέρθηκε προηγουμένως, ο έλεγχος ταυτότητας Microsoft Entra ID γίνεται αυτόματα κατά την είσοδό σας στον χώρο εργασίας Fabric, επομένως χρειάζεται να χρησιμοποιήσετε αυτές τις μεθόδους μόνο εάν το απαιτεί το σενάριό σας.

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)

from azure.identity import ClientSecretCredential
credential = ClientSecretCredential(tenant_id="", client_id="", client_secret="") # service principal app
scope = "https://database.windows.net/.default"
token = credential.get_token(scope).token

df.write.mode("overwrite").option("url", url).option("accesstoken", token).mssql("dbo.publicExample")
spark.read.option("accesstoken", token).mssql("dbo.publicExample").show()
spark.read.option("accesstoken", token).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

Υποστηριζόμενες λειτουργίες αποθήκευσης του DataFrame

Όταν γράφετε δεδομένα από βάσεις δεδομένων Spark σε βάσεις δεδομένων SQL, μπορείτε να επιλέξετε από πολλές λειτουργίες αποθήκευσης. Οι λειτουργίες αποθήκευσης ελέγχουν τον τρόπο εγγραφής των δεδομένων όταν υπάρχει ήδη ο πίνακας προορισμού και μπορούν να επηρεάσουν το σχήμα, τα δεδομένα και τη δημιουργία ευρετηρίου. Η κατανόηση αυτών των λειτουργιών σάς βοηθά να αποφύγετε απροσδόκητη απώλεια δεδομένων ή αλλαγές.

Αυτή η σύνδεση υποστηρίζει τις επιλογές που ορίζονται εδώ: Λειτουργίες Spark Save

  • ErrorIfExists (προεπιλεγμένη λειτουργία αποθήκευσης): Εάν υπάρχει ο πίνακας προορισμού, η εγγραφή ματαιώνεται και επιστρέφεται μια εξαίρεση. Διαφορετικά, δημιουργείται ένας νέος πίνακας με δεδομένα.

  • Παράβλεψη: Εάν υπάρχει ο πίνακας προορισμού, η εγγραφή αγνοεί το αίτημα και δεν επιστρέφει σφάλμα. Διαφορετικά, δημιουργείται ένας νέος πίνακας με δεδομένα.

  • Αντικατάσταση: Εάν υπάρχει ο πίνακας προορισμού, ο πίνακας απορρίπτεται, δημιουργείται εκ νέου και προσαρτώνται νέα δεδομένα.

    Σημείωμα

    Όταν χρησιμοποιείτε overwriteτο , χάνετε το αρχικό σχήμα πίνακα (ειδικά τους αποκλειστικούς τύπους δεδομένων MSSQL) και τους δείκτες πίνακα. Το σχήμα αντικαθίσταται από το σχήμα που συνάγεται από το Spark DataFrame. Για να αποφύγετε την απώλεια του σχήματος και των δεικτών, προσθέστε .option("truncate", true).

  • Προσάρτηση: Εάν υπάρχει ο πίνακας προορισμού, προστίθενται νέα δεδομένα σε αυτόν. Διαφορετικά, δημιουργείται ένας νέος πίνακας με δεδομένα.

Αντιμετώπιση προβλημάτων

Όταν ολοκληρωθεί η διαδικασία, η έξοδος της λειτουργίας ανάγνωσης Spark εμφανίζεται στην περιοχή εξόδου του κελιού. Τα σφάλματα προέρχονται απευθείας com.microsoft.sqlserver.jdbc.SQLServerException από τον SQL Server. Μπορείτε να βρείτε λεπτομερείς πληροφορίες σφαλμάτων στα αρχεία καταγραφής της εφαρμογής Spark.

Οι μαζικές εγγραφές απαιτούν εισερχόμενα δεδομένα για να ταιριάζουν στους τύπους δεδομένων SQL προορισμού. Εάν τα δεδομένα δεν χωράνε, ενδέχεται να λάβετε αυτό το σφάλμα:

Caused by: com.microsoft.sqlserver.jdbc.SQLServerException: The service has encountered an error processing your request. Please try again. Error code 4815.

Για παράδειγμα, αυτό το σφάλμα μπορεί να προκύψει όταν ο πίνακας SQL προορισμού χρησιμοποιεί datetime, αλλά μια εισερχόμενη τιμή Spark TimestampType , όπως 2025-01-01 10:30:00.123456, έχει μεγαλύτερη ακρίβεια από datetime τις υποστηρίζει.

Για να επιλύσετε το σφάλμα, χρησιμοποιήστε μία από αυτές τις προσεγγίσεις:

  • Μεταφέρετε, περικόψτε ή μετασχηματίστε τα εισερχόμενα δεδομένα ώστε να ταιριάζουν στον τύπο δεδομένων SQL προορισμού. Για παράδειγμα, περικόψτε την τιμή σε τρία ψηφία ακρίβειας κλασματικού δευτερολέπτου: 2025-01-01 10:30:00.123000.
  • Επιτρέψτε στη γραμμή σύνδεσης να δημιουργήσει ξανά το σχήμα πίνακα ορίζοντας ..option("truncate", false) Η σύνδεση αντιστοιχίζει τον τύπο χρονικής σήμανσης Spark σε SQL datetime2.

Σχετικό περιεχόμενο

  • Τελευταία ενημέρωση στις