Udostępnij przez

Łącznik dedykowanej puli SQL Azure Synapse dla platformy Apache Spark

Wprowadzenie

Łącznik dedykowanej puli SQL dla Apache Spark w usłudze Azure Synapse Analytics umożliwia efektywne przesyłanie dużych zestawów danych między środowiskiem uruchomieniowym Apache Spark a dedykowaną pulą SQL. Łącznik jest dostarczany jako biblioteka domyślna z obszarem roboczym usługi Azure Synapse. Łącznik jest implementowany przy użyciu Scala języka. Łącznik obsługuje języki Scala i Python. Aby użyć łącznika z innymi opcjami języka notesu, użyj komendy magicznej platformy Spark — %%spark.

Na wysokim poziomie łącznik zapewnia następujące możliwości:

  • Odczyt z dedykowanej puli SQL usługi Azure Synapse:
    • Odczytywanie dużych zestawów danych z tabel i widoków dedykowanej puli SQL (wewnętrznych i zewnętrznych) usługi Synapse.
    • Kompleksowa obsługa wypychania predykatów, gdzie filtry w DataFrame są mapowane na odpowiednie wypychanie predykatów SQL.
    • Obsługa oczyszczania kolumn.
    • Obsługa przenoszenia wykonania zapytania do niższego poziomu.
  • Zapisz w dedykowanej puli SQL usługi Azure Synapse:
    • Wczytywanie dużej ilości danych do tabel typu wewnętrznego i zewnętrznego.
    • Obsługuje następujące preferencje trybu zapisywania ramki danych:
      • Append
      • ErrorIfExists
      • Ignore
      • Overwrite
    • Typ zapisu w tabeli zewnętrznej obsługuje format pliku Parquet oraz rozdzielany tekst (na przykład CSV).
    • Aby teraz zapisywać dane w tabelach wewnętrznych, łącznik używa instrukcji COPY zamiast metody CETAS/CTAS.
    • Ulepszenia kompleksowej wydajności przepustowości zapisu.
    • Wprowadza opcjonalny uchwyt wywołania zwrotnego, który jest argumentem funkcji Scala, i którego klienci mogą używać do odbierania metryk po zapisie.
      • Oto kilka przykładów — liczba rekordów, czas trwania wykonania określonej akcji i przyczyna niepowodzenia.

Podejście orkiestracji

Przeczytaj

Diagram przepływu danych wysokiego poziomu opisujący koordynację łącznika dla żądania odczytu.

Napisz

Diagram przepływu danych wysokiego poziomu opisujący orkiestrację łącznika podczas obsługi żądania zapisu.

Wymagania wstępne

Wymagania wstępne, takie jak konfigurowanie wymaganych zasobów platformy Azure i kroki ich konfigurowania, zostały omówione w tej sekcji.

Zasoby platformy Azure

Przejrzyj i skonfiguruj następujące zależne zasoby platformy Azure:

Przygotowywanie bazy danych

Połącz się z dedykowaną bazą danych puli SQL usługi Synapse i uruchom następujące instrukcje konfiguracji:

  • Utwórz użytkownika bazy danych, który jest mapowany na tożsamość użytkownika Microsoft Entra, używaną do logowania się do obszaru roboczego usługi Azure Synapse.

    CREATE USER [username@domain.com] FROM EXTERNAL PROVIDER;

  • Utwórz schemat, w którym zostaną zdefiniowane tabele, tak aby łącznik mógł pomyślnie zapisywać i odczytywać z odpowiednich tabel.

    CREATE SCHEMA [<schema_name>];

Uwierzytelnianie

Uwierzytelnianie oparte na identyfikatorze Entra firmy Microsoft

Uwierzytelnianie oparte na identyfikatorze Entra firmy Microsoft to zintegrowane podejście do uwierzytelniania. Użytkownik musi pomyślnie zalogować się do obszaru roboczego usługi Azure Synapse Analytics.

Uwierzytelnianie podstawowe

Podstawowe podejście do uwierzytelniania wymaga skonfigurowania username i password opcji przez użytkownika. Zapoznaj się z sekcją — Opcje konfiguracji, aby dowiedzieć się więcej o odpowiednich parametrach konfiguracji do odczytu z tabel i zapisywania ich w tabelach w dedykowanej puli SQL usługi Azure Synapse.

Autoryzacja

Azure Data Lake Storage Gen2

Istnieją dwa sposoby udzielania uprawnień dostępu do usługi Azure Data Lake Storage Gen2 — konto magazynu:

  • Rola oparta na kontroli dostępu — Rola współautora danych obiektu blob w usłudze Storage
    • Przypisanie Storage Blob Data Contributor Role przyznaje użytkownikowi uprawnienia do odczytu, zapisu i usuwania z kontenerów obiektów blob w usłudze Azure Storage.
    • Kontrola dostępu oparta na rolach oferuje ogólne podejście do kontroli na poziomie kontenera.
  • Listy kontroli dostępu (ACL)
    • Podejście ACL umożliwia precyzyjne sterowanie określonymi ścieżkami i/lub plikami w danym folderze.
    • Sprawdzanie listy kontroli dostępu (ACL) nie jest wymuszane, jeśli użytkownik ma już przyznane uprawnienia w ramach podejścia opartego na rolach (RBAC).
    • Istnieją dwa główne rodzaje uprawnień listy ACL.
      • Uprawnienia dostępu (stosowane na określonym poziomie lub obiekcie).
      • Uprawnienia domyślne (automatycznie stosowane dla wszystkich obiektów podrzędnych w momencie ich tworzenia).
    • Rodzaje uprawnień obejmują:
      • Execute umożliwia przechodzenie do hierarchii folderów lub nawigowanie po niej.
      • Read umożliwia odczytywanie.
      • Write umożliwia zapisywanie.
    • Należy skonfigurować listy kontroli dostępu ACL, tak aby łącznik mógł pomyślnie zapisywać i odczytywać z lokalizacji magazynowych.

Uwaga

  • Jeśli chcesz uruchamiać notesy przy użyciu potoków obszaru roboczego usługi Synapse, musisz również przyznać wymienione powyżej uprawnienia dostępu do zarządzanej tożsamości domyślnej Synapse Workspace. Nazwa domyślnej tożsamości zarządzanej obszaru roboczego jest taka sama jak nazwa obszaru roboczego.

  • Aby używać obszaru roboczego usługi Synapse z zabezpieczonymi kontami magazynu, należy skonfigurować zarządzany prywatny punkt końcowy z poziomu notesu. Aby zatwierdzić zarządzany prywatny punkt końcowy, należy to zrobić z sekcji Private endpoint connections na koncie magazynu usługi ADLS Gen2 w okienku Networking.

Dedykowana pula SQL w usłudze Azure Synapse

Aby umożliwić udaną interakcję z dedykowaną pulą SQL w usłudze Azure Synapse, konieczna jest następująca autoryzacja, chyba że użytkownik jest również skonfigurowany jako Active Directory Admin w dedykowanym punkcie końcowym SQL.

  • Przeczytaj scenariusz

    • Udziel użytkownikowi db_exporter przy użyciu procedury składowanej systemu sp_addrolemember.

      EXEC sp_addrolemember 'db_exporter', [<your_domain_user>@<your_domain_name>.com];

  • Napisz scenariusz

    • Konektor używa polecenia COPY do zapisywania danych ze strefy tymczasowej do zarządzanej lokalizacji tabeli wewnętrznej.

      • Skonfiguruj wymagane uprawnienia opisane tutaj.

      • Poniżej znajduje się fragment kodu szybkiego dostępu:

        --Make sure your user has the permissions to CREATE tables in the [dbo] schema
GRANT CREATE TABLE TO [<your_domain_user>@<your_domain_name>.com];
GRANT ALTER ON SCHEMA::<target_database_schema_name> TO [<your_domain_user>@<your_domain_name>.com];

--Make sure your user has ADMINISTER DATABASE BULK OPERATIONS permissions
GRANT ADMINISTER DATABASE BULK OPERATIONS TO [<your_domain_user>@<your_domain_name>.com];

--Make sure your user has INSERT permissions on the target table
GRANT INSERT ON <your_table> TO [<your_domain_user>@<your_domain_name>.com]

Dokumentacja interfejsu API

Łącznik dedykowanej puli SQL usługi Azure Synapse dla platformy Apache Spark — dokumentacja interfejsu API.

Opcje konfiguracji

Aby pomyślnie uruchomić i zorganizować operację odczytu lub zapisu, łącznik oczekuje pewnych parametrów konfiguracji. Definicja obiektu — com.microsoft.spark.sqlanalytics.utils.Constants zawiera listę ustandaryzowanych stałych dla każdego klucza parametru.

Poniżej znajduje się lista opcji konfiguracji opartych na scenariuszu użycia:

  • Odczyt przy użyciu uwierzytelniania opartego na identyfikatorze Entra firmy Microsoft
    • Poświadczenia są automatycznie mapowane, a użytkownik nie musi udostępniać określonych opcji konfiguracji.
    • Trzyczęściowy argument nazwy tabeli w synapsesql metodzie jest wymagany do odczytu z odpowiedniej tabeli w dedykowanej puli SQL usługi Azure Synapse.
  • Odczyt przy użyciu uwierzytelniania podstawowego
    • Dedykowany punkt końcowy SQL usługi Azure Synapse
      • Constants.SERVER — Punkt końcowy dedykowanej puli SQL Synapse (FQDN serwera)
      • Constants.USER — Nazwa użytkownika SQL.
      • Constants.PASSWORD — Hasło użytkownika SQL.
    • Punkt końcowy usługi Azure Data Lake Storage (Gen 2) — foldery przejściowe
      • Constants.DATA_SOURCE — Ścieżka magazynu ustawiona w parametrze lokalizacji źródła danych jest używana do etapu przygotowania danych.
  • Pisanie przy użyciu uwierzytelniania opartego na identyfikatorze Entra firmy Microsoft
    • Dedykowany punkt końcowy SQL usługi Azure Synapse
      • Domyślnie konektor wywnioskuje punkt końcowy Synapse Dedicated SQL przy użyciu nazwy bazy danych ustawionej na parametrze nazwy tabeli w trójczłonowej notacji metody synapsesql.
      • Alternatywnie użytkownicy mogą użyć Constants.SERVER opcji , aby określić punkt końcowy sql. Upewnij się, że punkt końcowy hostuje odpowiednią bazę danych z odpowiednim schematem.
    • Punkt końcowy usługi Azure Data Lake Storage (Gen 2) — foldery przejściowe
      • W przypadku typu wewnętrznej tabeli:
        • Skonfiguruj opcję Constants.TEMP_FOLDER lub .Constants.DATA_SOURCE
        • Jeśli użytkownik zdecydował się podać opcję Constants.DATA_SOURCE, folder przejściowy będzie uzyskiwany przy użyciu wartości location ze źródła danych.
        • Jeśli podano obie wartości, zostanie użyta Constants.TEMP_FOLDER wartość opcji.
        • W przypadku braku opcji folderu przejściowego łącznik będzie ją tworzyć na podstawie konfiguracji w trakcie działania — spark.sqlanalyticsconnector.stagingdir.prefix.
      • W przypadku typu tabeli zewnętrznej:
        • Constants.DATA_SOURCE jest wymaganą opcją konfiguracji.
        • Łącznik używa ścieżki magazynu ustawionej w parametrze lokalizacji źródła danych w połączeniu z argumentem location do metody synapsesql, aby tworzyć ścieżkę bezwzględną do utrwalania danych tabeli zewnętrznej.
        • location Jeśli argument metody synapsesql nie zostanie określony, łącznik będzie uzyskiwać wartość lokalizacji jako <base_path>/dbName/schemaName/tableName.
  • Pisanie przy użyciu uwierzytelniania podstawowego
    • Dedykowany punkt końcowy SQL usługi Azure Synapse
      • Constants.SERVER - — Dedykowany punkt końcowy puli SQL usługi Synapse (nazwa FQDN serwera).
      • Constants.USER — Nazwa użytkownika SQL.
      • Constants.PASSWORD — Hasło użytkownika SQL.
      • Constants.STAGING_STORAGE_ACCOUNT_KEY skojarzone z kontem magazynu hostującym Constants.TEMP_FOLDERS (tylko dla wewnętrznych typów tabel) lub Constants.DATA_SOURCE.
    • Punkt końcowy usługi Azure Data Lake Storage (Gen 2) — foldery przejściowe
      • Poświadczenia podstawowego uwierzytelniania SQL nie mają zastosowania do uzyskiwania dostępu do punktów końcowych magazynu.
      • W związku z tym należy przypisać odpowiednie uprawnienia dostępu do magazynu zgodnie z opisem w sekcji Azure Data Lake Storage Gen2.

Szablony kodu

W tej sekcji przedstawiono szablony kodu referencyjnego opisujące sposób używania i wywoływania dedykowanej puli SQL usługi Azure Synapse dla platformy Apache Spark.

Uwaga

Korzystanie z łącznika w języku Python—

Odczyt z dedykowanej puli SQL usługi Azure Synapse

Żądanie odczytu — synapsesql sygnatura metody

synapsesql(tableName:String="") => org.apache.spark.sql.DataFrame

Odczytywanie z tabeli przy użyciu uwierzytelniania opartego na identyfikatorze Microsoft Entra

//Use case is to read data from an internal table in Synapse Dedicated SQL Pool DB
//Azure Active Directory based authentication approach is preferred here.
import org.apache.spark.sql.DataFrame
import com.microsoft.spark.sqlanalytics.utils.Constants
import org.apache.spark.sql.SqlAnalyticsConnector._

//Read from existing internal table
val dfToReadFromTable:DataFrame = spark.read.
    //If `Constants.SERVER` is not provided, the `<database_name>` from the three-part table name argument 
    //to `synapsesql` method is used to infer the Synapse Dedicated SQL End Point.
    option(Constants.SERVER, "<sql-server-name>.sql.azuresynapse.net").
    //Defaults to storage path defined in the runtime configurations
    option(Constants.TEMP_FOLDER, "abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<some_base_path_for_temporary_staging_folders>").
    //Three-part table name from where data will be read.
    synapsesql("<database_name>.<schema_name>.<table_name>").
    //Column-pruning i.e., query select column values.
    select("<some_column_1>", "<some_column_5>", "<some_column_n>"). 
    //Push-down filter criteria that gets translated to SQL Push-down Predicates.    
    filter(col("Title").startsWith("E")).
    //Fetch a sample of 10 records 
    limit(10)

//Show contents of the dataframe
dfToReadFromTable.show()

Odczytywanie z zapytania przy użyciu uwierzytelniania opartego na identyfikatorze Entra firmy Microsoft

Uwaga

Ograniczenia podczas odczytywania z zapytania:

  • Nie można jednocześnie określić nazwy tabeli i kwerendy.
  • Dozwolone są tylko wybrane zapytania. Komendy DDL i DML SQL są niedozwolone.
  • Opcje wybierania i filtrowania w ramce danych nie są wypychane do dedykowanej puli SQL po określeniu zapytania.
  • Odczyt z zapytania jest dostępny tylko na platformie Spark 3.
//Use case is to read data from an internal table in Synapse Dedicated SQL Pool DB
//Azure Active Directory based authentication approach is preferred here.
import org.apache.spark.sql.DataFrame
import com.microsoft.spark.sqlanalytics.utils.Constants
import org.apache.spark.sql.SqlAnalyticsConnector._


// Read from a query
// Query can be provided either as an argument to synapsesql or as a Constant - Constants.QUERY
val dfToReadFromQueryAsOption:DataFrame = spark.read.
    // Name of the SQL Dedicated Pool or database where to run the query
    // Database can be specified as a Spark Config - spark.sqlanalyticsconnector.dw.database or as a Constant - Constants.DATABASE
     option(Constants.DATABASE, "<database_name>").
    //If `Constants.SERVER` is not provided, the `<database_name>` from the three-part table name argument 
    //to `synapsesql` method is used to infer the Synapse Dedicated SQL End Point.
    option(Constants.SERVER, "<sql-server-name>.sql.azuresynapse.net").
    //Defaults to storage path defined in the runtime configurations
    option(Constants.TEMP_FOLDER, "abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<some_base_path_for_temporary_staging_folders>")
    //query from which data will be read
    .option(Constants.QUERY, "select <column_name>, count(*) as cnt from <schema_name>.<table_name> group by <column_name>")
    synapsesql()

val dfToReadFromQueryAsArgument:DataFrame = spark.read.
     // Name of the SQL Dedicated Pool or database where to run the query
     // Database can be specified as a Spark Config - spark.sqlanalyticsconnector.dw.database or as a Constant - Constants.DATABASE
     option(Constants.DATABASE, "<database_name>")
    //If `Constants.SERVER` is not provided, the `<database_name>` from the three-part table name argument 
    //to `synapsesql` method is used to infer the Synapse Dedicated SQL End Point.
    option(Constants.SERVER, "<sql-server-name>.sql.azuresynapse.net").
    //Defaults to storage path defined in the runtime configurations
    option(Constants.TEMP_FOLDER, "abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<some_base_path_for_temporary_staging_folders>")
    //query from which data will be read
    .synapsesql("select <column_name>, count(*) as counts from <schema_name>.<table_name> group by <column_name>")


//Show contents of the dataframe
dfToReadFromQueryAsOption.show()
dfToReadFromQueryAsArgument.show()

Odczytywanie z tabeli przy użyciu uwierzytelniania podstawowego

//Use case is to read data from an internal table in Synapse Dedicated SQL Pool DB
//Azure Active Directory based authentication approach is preferred here.
import org.apache.spark.sql.DataFrame
import com.microsoft.spark.sqlanalytics.utils.Constants
import org.apache.spark.sql.SqlAnalyticsConnector._

//Read from existing internal table
val dfToReadFromTable:DataFrame = spark.read.
    //If `Constants.SERVER` is not provided, the `<database_name>` from the three-part table name argument 
    //to `synapsesql` method is used to infer the Synapse Dedicated SQL End Point.
    option(Constants.SERVER, "<sql-server-name>.sql.azuresynapse.net").
    //Set database user name
    option(Constants.USER, "<user_name>").
    //Set user's password to the database
    option(Constants.PASSWORD, "<user_password>").
    //Set name of the data source definition that is defined with database scoped credentials.
    //Data extracted from the table will be staged to the storage path defined on the data source's location setting.
    option(Constants.DATA_SOURCE, "<data_source_name>").
    //Three-part table name from where data will be read.
    synapsesql("<database_name>.<schema_name>.<table_name>").
    //Column-pruning i.e., query select column values.
    select("<some_column_1>", "<some_column_5>", "<some_column_n>"). 
    //Push-down filter criteria that gets translated to SQL Push-down Predicates.    
    filter(col("Title").startsWith("E")).
    //Fetch a sample of 10 records 
    limit(10)
    

//Show contents of the dataframe
dfToReadFromTable.show()

Odczytywanie z zapytania przy użyciu uwierzytelniania podstawowego

//Use case is to read data from an internal table in Synapse Dedicated SQL Pool DB
//Azure Active Directory based authentication approach is preferred here.
import org.apache.spark.sql.DataFrame
import com.microsoft.spark.sqlanalytics.utils.Constants
import org.apache.spark.sql.SqlAnalyticsConnector._

// Name of the SQL Dedicated Pool or database where to run the query
// Database can be specified as a Spark Config or as a Constant - Constants.DATABASE
spark.conf.set("spark.sqlanalyticsconnector.dw.database", "<database_name>")

// Read from a query
// Query can be provided either as an argument to synapsesql or as a Constant - Constants.QUERY
val dfToReadFromQueryAsOption:DataFrame = spark.read.
     //Name of the SQL Dedicated Pool or database where to run the query
     //Database can be specified as a Spark Config - spark.sqlanalyticsconnector.dw.database or as a Constant - Constants.DATABASE
      option(Constants.DATABASE, "<database_name>").
    //If `Constants.SERVER` is not provided, the `<database_name>` from the three-part table name argument 
    //to `synapsesql` method is used to infer the Synapse Dedicated SQL End Point.
    option(Constants.SERVER, "<sql-server-name>.sql.azuresynapse.net").
    //Set database user name
    option(Constants.USER, "<user_name>").
    //Set user's password to the database
    option(Constants.PASSWORD, "<user_password>").
    //Set name of the data source definition that is defined with database scoped credentials.
    //Data extracted from the SQL query will be staged to the storage path defined on the data source's location setting.
    option(Constants.DATA_SOURCE, "<data_source_name>").
    //Query where data will be read.  
    option(Constants.QUERY, "select <column_name>, count(*) as counts from <schema_name>.<table_name> group by <column_name>" ).
    synapsesql()

val dfToReadFromQueryAsArgument:DataFrame = spark.read.
     //Name of the SQL Dedicated Pool or database where to run the query
     //Database can be specified as a Spark Config - spark.sqlanalyticsconnector.dw.database or as a Constant - Constants.DATABASE
      option(Constants.DATABASE, "<database_name>").
    //If `Constants.SERVER` is not provided, the `<database_name>` from the three-part table name argument 
    //to `synapsesql` method is used to infer the Synapse Dedicated SQL End Point.
    option(Constants.SERVER, "<sql-server-name>.sql.azuresynapse.net").
    //Set database user name
    option(Constants.USER, "<user_name>").
    //Set user's password to the database
    option(Constants.PASSWORD, "<user_password>").
    //Set name of the data source definition that is defined with database scoped credentials.
    //Data extracted from the SQL query will be staged to the storage path defined on the data source's location setting.
    option(Constants.DATA_SOURCE, "<data_source_name>").
    //Query where data will be read.  
    synapsesql("select <column_name>, count(*) as counts from <schema_name>.<table_name> group by <column_name>")
    

//Show contents of the dataframe
dfToReadFromQueryAsOption.show()
dfToReadFromQueryAsArgument.show()

Napisz do dedykowanej puli SQL usługi Azure Synapse

Żądanie zapisu — synapsesql sygnatura metody

synapsesql(tableName:String, 
           tableType:String = Constants.INTERNAL, 
           location:Option[String] = None,
           callBackHandle=Option[(Map[String, Any], Option[Throwable])=>Unit]):Unit

Pisanie przy użyciu uwierzytelniania opartego na identyfikatorze Entra firmy Microsoft

Poniżej znajduje się kompleksowy szablon kodu opisujący sposób używania Connectora do scenariuszy zapisu:

//Add required imports
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.SaveMode
import com.microsoft.spark.sqlanalytics.utils.Constants
import org.apache.spark.sql.SqlAnalyticsConnector._

//Define read options for example, if reading from CSV source, configure header and delimiter options.
val pathToInputSource="abfss://<storage_container_name>@<storage_account_name>.dfs.core.windows.net/<some_folder>/<some_dataset>.csv"

//Define read configuration for the input CSV
val dfReadOptions:Map[String, String] = Map("header" -> "true", "delimiter" -> ",")

//Initialize DataFrame that reads CSV data from a given source 
val readDF:DataFrame=spark.
            read.
            options(dfReadOptions).
            csv(pathToInputSource).
            limit(1000) //Reads first 1000 rows from the source CSV input.

//Setup and trigger the read DataFrame for write to Synapse Dedicated SQL Pool.
//Fully qualified SQL Server DNS name can be obtained using one of the following methods:
//    1. Synapse Workspace - Manage Pane - SQL Pools - <Properties view of the corresponding Dedicated SQL Pool>
//    2. From Azure Portal, follow the bread-crumbs for <Portal_Home> -> <Resource_Group> -> <Dedicated SQL Pool> and then go to Connection Strings/JDBC tab. 
//If `Constants.SERVER` is not provided, the value will be inferred by using the `database_name` in the three-part table name argument to the `synapsesql` method.
//Like-wise, if `Constants.TEMP_FOLDER` is not provided, the connector will use the runtime staging directory config (see section on Configuration Options for details).
val writeOptionsWithAADAuth:Map[String, String] = Map(Constants.SERVER -> "<dedicated-pool-sql-server-name>.sql.azuresynapse.net",
                                            Constants.TEMP_FOLDER -> "abfss://<storage_container_name>@<storage_account_name>.dfs.core.windows.net/<some_temp_folder>")

//Setup optional callback/feedback function that can receive post write metrics of the job performed.
var errorDuringWrite:Option[Throwable] = None
val callBackFunctionToReceivePostWriteMetrics: (Map[String, Any], Option[Throwable]) => Unit =
    (feedback: Map[String, Any], errorState: Option[Throwable]) => {
    println(s"Feedback map - ${feedback.map{case(key, value) => s"$key -> $value"}.mkString("{",",\n","}")}")
    errorDuringWrite = errorState
}

//Configure and submit the request to write to Synapse Dedicated SQL Pool (note - default SaveMode is set to ErrorIfExists)
//Sample below is using AAD-based authentication approach; See further examples to leverage SQL Basic auth.
readDF.
    write.
    //Configure required configurations.
    options(writeOptionsWithAADAuth).
    //Choose a save mode that is apt for your use case.
    mode(SaveMode.Overwrite).
    synapsesql(tableName = "<database_name>.<schema_name>.<table_name>", 
                //For external table type value is Constants.EXTERNAL
                tableType = Constants.INTERNAL, 
                //Optional parameter that is used to specify external table's base folder; defaults to `database_name/schema_name/table_name`
                location = None, 
                //Optional parameter to receive a callback.
                callBackHandle = Some(callBackFunctionToReceivePostWriteMetrics))

//If write request has failed, raise an error and fail the Cell's execution.
if(errorDuringWrite.isDefined) throw errorDuringWrite.get

Pisanie przy użyciu uwierzytelniania podstawowego

Poniższy fragment kodu zastępuje definicję zapisu opisaną w sekcji Pisanie przy użyciu uwierzytelniania opartego na identyfikatorze Entra firmy Microsoft, aby przesłać żądanie zapisu przy użyciu podstawowego podejścia do uwierzytelniania SQL:

//Define write options to use SQL basic authentication
val writeOptionsWithBasicAuth:Map[String, String] = Map(Constants.SERVER -> "<dedicated-pool-sql-server-name>.sql.azuresynapse.net",
                                           //Set database user name
                                           Constants.USER -> "<user_name>",
                                           //Set database user's password
                                           Constants.PASSWORD -> "<user_password>",
                                           //Required only when writing to an external table. For write to internal table, this can be used instead of TEMP_FOLDER option.
                                           Constants.DATA_SOURCE -> "<Name of the datasource as defined in the target database>"
                                           //To be used only when writing to internal tables. Storage path will be used for data staging.
                                           Constants.TEMP_FOLDER -> "abfss://<storage_container_name>@<storage_account_name>.dfs.core.windows.net/<some_temp_folder>")

//Configure and submit the request to write to Synapse Dedicated SQL Pool. 
readDF.
    write.
    options(writeOptionsWithBasicAuth).
    //Choose a save mode that is apt for your use case.
    mode(SaveMode.Overwrite). 
    synapsesql(tableName = "<database_name>.<schema_name>.<table_name>", 
                //For external table type value is Constants.EXTERNAL
                tableType = Constants.INTERNAL,
                //Not required for writing to an internal table 
                location = None,
                //Optional parameter.
                callBackHandle = Some(callBackFunctionToReceivePostWriteMetrics))

W podstawowym podejściu do uwierzytelniania, aby odczytać dane ze źródłowej ścieżki magazynu, wymagane są inne opcje konfiguracji. Poniższy fragment kodu zawiera przykład odczytu ze źródła danych usługi Azure Data Lake Storage Gen2 przy użyciu poświadczeń jednostki usługi:

//Specify options that Spark runtime must support when interfacing and consuming source data
val storageAccountName="<storageAccountName>"
val storageContainerName="<storageContainerName>"
val subscriptionId="<AzureSubscriptionID>"
val spnClientId="<ServicePrincipalClientID>"
val spnSecretKeyUsedAsAuthCred="<spn_secret_key_value>"
val dfReadOptions:Map[String, String]=Map("header"->"true",
                                "delimiter"->",", 
                                "fs.defaultFS" -> s"abfss://$storageContainerName@$storageAccountName.dfs.core.windows.net",
                                s"fs.azure.account.auth.type.$storageAccountName.dfs.core.windows.net" -> "OAuth",
                                s"fs.azure.account.oauth.provider.type.$storageAccountName.dfs.core.windows.net" -> 
                                    "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
                                "fs.azure.account.oauth2.client.id" -> s"$spnClientId",
                                "fs.azure.account.oauth2.client.secret" -> s"$spnSecretKeyUsedAsAuthCred",
                                "fs.azure.account.oauth2.client.endpoint" -> s"https://login.microsoftonline.com/$subscriptionId/oauth2/token",
                                "fs.AbstractFileSystem.abfss.impl" -> "org.apache.hadoop.fs.azurebfs.Abfs",
                                "fs.abfss.impl" -> "org.apache.hadoop.fs.azurebfs.SecureAzureBlobFileSystem")
//Initialize the Storage Path string, where source data is maintained/kept.
val pathToInputSource=s"abfss://$storageContainerName@$storageAccountName.dfs.core.windows.net/<base_path_for_source_data>/<specific_file (or) collection_of_files>"
//Define data frame to interface with the data source
val df:DataFrame = spark.
            read.
            options(dfReadOptions).
            csv(pathToInputSource).
            limit(100)

Obsługiwane tryby zapisywania ramki danych

Następujące tryby zapisywania są obsługiwane podczas zapisywania danych źródłowych w tabeli docelowej w dedykowanej puli SQL usługi Azure Synapse:

  • ErrorIfExists (domyślny tryb zapisywania)
    • Jeśli tabela docelowa istnieje, zapis zostanie przerwany, a do wywołującego zostanie zwrócony wyjątek. W przeciwnym razie zostanie utworzona nowa tabela z danymi z folderów przejściowych.
  • Ignorować
    • Jeśli tabela docelowa istnieje, operacja zapisania zignoruje żądanie bez zwracania błędu. W przeciwnym razie zostanie utworzona nowa tabela z danymi z folderów przejściowych.
  • Nadpisać
    • Jeśli tabela docelowa istnieje, istniejące dane w miejscu docelowym zostaną zastąpione danymi z folderów przejściowych. W przeciwnym razie zostanie utworzona nowa tabela z danymi z folderów przejściowych.
  • Dołączyć
    • Jeśli tabela docelowa istnieje, nowe dane zostaną do niej dołączone. W przeciwnym razie zostanie utworzona nowa tabela z danymi z folderów przejściowych.

Obsługa wywołania zwrotnego żądań zapisu

Zmiany wprowadzające nowe funkcje do API ścieżki zapisu wprowadziły eksperymentalną funkcję umożliwiającą klientowi dostęp do mapy klucz-wartość metryk po zapisie. Klucze metryk są definiowane w nowej definicji obiektu — Constants.FeedbackConstants. Metryki można pobrać jako ciąg JSON, przekazując uchwyt wywołania zwrotnego (a Scala Function). Poniżej znajduje się sygnatura funkcji:

//Function signature is expected to have two arguments - a `scala.collection.immutable.Map[String, Any]` and an Option[Throwable]
//Post-write if there's a reference of this handle passed to the `synapsesql` signature, it will be invoked by the closing process.
//These arguments will have valid objects in either Success or Failure case. In case of Failure the second argument will be a `Some(Throwable)`.
(Map[String, Any], Option[Throwable]) => Unit

Poniżej przedstawiono kilka godnych uwagi metryk (przedstawionych w przypadku camel):

  • WriteFailureCause
  • DataStagingSparkJobDurationInMilliseconds
  • NumberOfRecordsStagedForSQLCommit
  • SQLStatementExecutionDurationInMilliseconds
  • rows_processed

Poniżej znajduje się przykładowy ciąg JSON z metrykami po zapisie.

{
 SparkApplicationId -> <spark_yarn_application_id>,
 SQLStatementExecutionDurationInMilliseconds -> 10113,
 WriteRequestReceivedAtEPOCH -> 1647523790633,
 WriteRequestProcessedAtEPOCH -> 1647523808379,
 StagingDataFileSystemCheckDurationInMilliseconds -> 60,
 command -> "COPY INTO [schema_name].[table_name] ...",
 NumberOfRecordsStagedForSQLCommit -> 100,
 DataStagingSparkJobEndedAtEPOCH -> 1647523797245,
 SchemaInferenceAssertionCompletedAtEPOCH -> 1647523790920,
 DataStagingSparkJobDurationInMilliseconds -> 5252,
 rows_processed -> 100,
 SaveModeApplied -> TRUNCATE_COPY,
 DurationInMillisecondsToValidateFileFormat -> 75,
 status -> Completed,
 SparkApplicationName -> <spark_application_name>,
 ThreePartFullyQualifiedTargetTableName -> <database_name>.<schema_name>.<table_name>,
 request_id -> <query_id_as_retrieved_from_synapse_dedicated_sql_db_query_reference>,
 StagingFolderConfigurationCheckDurationInMilliseconds -> 2,
 JDBCConfigurationsSetupAtEPOCH -> 193,
 StagingFolderConfigurationCheckCompletedAtEPOCH -> 1647523791012,
 FileFormatValidationsCompletedAtEPOCHTime -> 1647523790995,
 SchemaInferenceCheckDurationInMilliseconds -> 91,
 SaveModeRequested -> Overwrite,
 DataStagingSparkJobStartedAtEPOCH -> 1647523791993,
 DurationInMillisecondsTakenToGenerateWriteSQLStatements -> 4
}

Więcej przykładów kodu

Używanie zmaterializowanych danych pomiędzy komórkami

Ramki danych Spark można używać do uzyskiwania dostępu do danych pobranych w innej komórce, rejestrując widok tymczasowy.

  • Komórka, skąd pobierane są dane (np. z ustawionym językiem dla notesu jako Scala)
    //Necessary imports
    import org.apache.spark.sql.DataFrame
    import org.apache.spark.sql.SaveMode
    import com.microsoft.spark.sqlanalytics.utils.Constants
    import org.apache.spark.sql.SqlAnalyticsConnector._
    
    //Configure options and read from Synapse Dedicated SQL Pool.
    val readDF = spark.read.
        //Set Synapse Dedicated SQL End Point name.
        option(Constants.SERVER, "<synapse-dedicated-sql-end-point>.sql.azuresynapse.net").
        //Set database user name.
        option(Constants.USER, "<user_name>").
        //Set database user's password. 
        option(Constants.PASSWORD, "<user_password>").
        //Set name of the data source definition that is defined with database scoped credentials.
        option(Constants.DATA_SOURCE,"<data_source_name>").
        //Set the three-part table name from which the read must be performed.
        synapsesql("<database_name>.<schema_name>.<table_name>").
        //Optional - specify number of records the DataFrame would read.
        limit(10)
    //Register the temporary view (scope - current active Spark Session)
    readDF.createOrReplaceTempView("<temporary_view_name>")
  • Teraz zmień preferencje językowe notebooka na PySpark (Python) i pobierz dane z zarejestrowanego widoku <temporary_view_name>.
        spark.sql("select * from <temporary_view_name>").show()

Zarządzanie odpowiedziami

Wywoływanie synapsesql ma dwa możliwe stany końcowe — powodzenie lub stan niepowodzenia. W tej sekcji opisano sposób obsługi odpowiedzi na żądanie dla każdego scenariusza.

Odczytaj odpowiedź na żądanie

Po zakończeniu fragment odpowiedzi odczytu jest wyświetlany w danych wyjściowych komórki. Błąd w bieżącej komórce spowoduje również anulowanie wykonania kolejnych komórek. Szczegółowe informacje o błędzie są dostępne w dziennikach aplikacji platformy Spark.

Napisz odpowiedź na żądanie

Domyślnie odpowiedź zapisu jest drukowana w danych wyjściowych komórki. Po awarii bieżąca komórka jest oznaczona jako uszkodzona, a kolejne wykonania komórek zostaną wstrzymane. Innym podejściem jest przekazanie opcji uchwytu wywołania zwrotnego do synapsesql metody. Uchwyt wywołania zwrotnego zapewni dostęp programistyczny do odpowiedzi zapisu.

Inne uwagi

  • Podczas odczytywania z tabel dedykowanej puli SQL usługi Azure Synapse:
    • Rozważ zastosowanie niezbędnych filtrów w ramce danych, aby skorzystać z funkcji oczyszczania kolumn łącznika.
    • Scenariusz odczytu nie obsługuje klauzuli TOP(n-rows) podczas formułowania instrukcji zapytania SELECT. Możliwość ograniczenia danych polega na użyciu klauzuli limit(.) ramki danych.
  • Podczas zapisywania danych w tabelach dedykowanej puli SQL usługi Azure Synapse:
    • W przypadku typów tabel wewnętrznych:
      • Tabele są tworzone przy użyciu dystrybucji danych ROUND_ROBIN.
      • Typy kolumn są wnioskowane z ramki danych, która odczytuje dane ze źródła. Kolumny typu string są mapowane na NVARCHAR(4000).
    • W przypadku typów tabel zewnętrznych:
      • Początkowy paralelizm ramki danych determinuje organizację danych dla tabeli zewnętrznej.
      • Typy kolumn są wnioskowane z ramki danych, która odczytuje dane ze źródła.
    • Lepszą dystrybucję danych między funkcjami wykonawczych można osiągnąć przez dostrajanie parametru spark.sql.files.maxPartitionBytes i ramki repartition danych.
    • Podczas pisania dużych zestawów danych ważne jest, aby uwzględnić wpływ ustawienia poziomu wydajności jednostek DWU, które ogranicza rozmiar transakcji.
  • Monitorowanie trendów wykorzystania usługi Azure Data Lake Storage Gen2 w celu wykrycia zachowań ograniczania przepustowości, które mogą mieć wpływ na wydajność odczytu i zapisu.

Powiązana zawartość

  • Last updated on