Aracılığıyla paylaş

Apache Spark için Azure Synapse Ayrılmış SQL Havuzu Bağlayıcısı

Giriş

Azure Synapse Analytics'teki Apache Spark için Azure Synapse Ayrılmış SQL Havuzu Bağlayıcısı, Apache Spark çalışma zamanı ile Ayrılmış SQL havuzu arasında büyük veri kümelerinin verimli bir şekilde aktarılmasını sağlar. Bağlayıcı, Azure Synapse Çalışma Alanı ile varsayılan kitaplık olarak gönderilir. Bağlayıcı, Scala dili kullanılarak uygulanır. Bağlayıcı Scala ve Python'ı destekler. Bağlayıcı'yı diğer not defteri dil seçenekleriyle kullanmak için Spark magic komutunu kullanın: %%spark.

Bağlayıcı üst düzeyde aşağıdaki özellikleri sağlar:

  • Azure Synapse Ayrılmış SQL Havuzu'ndan okuyun:
    • Synapse Özel SQL havuzu tablolarından (iç ve dış) ve görünümlerden büyük veri kümelerini okuyun.
    • DataFrame'deki filtrelerin ilgili SQL koşullarına eşlendiği kapsamlı koşul iteleme desteği.
    • Sütun budama desteği.
    • Sorgu aşağı gönderme desteği.
  • Azure Synapse Ayrılmış SQL Havuzu'na yazma:
    • Büyük hacimli verileri İç ve Dış tablo türlerine yükleyin.
    • Aşağıdaki DataFrame kaydetme modu tercihlerini destekler:
      • Append
      • ErrorIfExists
      • Ignore
      • Overwrite
    • Dış Tablo Yazma Türü, Parquet ve Ayrıştırılmış Metin dosya biçimlerini (örn. - CSV) destekler.
    • İç tablolara veri yazmak için bağlayıcı artık CETAS/CTAS yaklaşımı yerine COPY deyimini kullanıyor.
    • Uçtan uca yazma aktarım hızı performansını iyileştirmeye yönelik geliştirmeler.
    • Müşterilerin yazma sonrası metrikleri almak için kullanabileceği isteğe bağlı bir geri arama işlevini (Scala işlev bağımsız değişkeni) tanıtır.
      • Birkaç örnek şunlardır: kayıt sayısı, belirli bir eylemi tamamlama süresi ve hata nedeni.

Düzenleme yaklaşımı

Okundu

Bağlayıcının bir okuma isteğini düzenlemesini açıklayan üst düzey veri akışı diyagramı.

Yaz

Bağlayıcının yazma isteğini düzenlemesini açıklayan üst düzey veri akışı diyagramı.

Önkoşullar

Gerekli Azure kaynaklarını ayarlama gibi önkoşullar ve bunları yapılandırma adımları bu bölümde açıklanmıştır.

Azure kaynakları

Bağımlı Azure Kaynaklarını gözden geçirin ve ayarlayın:

Veritabanını hazırlama

Synapse Ayrılmış SQL Havuzu veritabanına bağlanın ve aşağıdaki kurulum deyimlerini çalıştırın:

  • Azure Synapse Çalışma Alanı'nda oturum açmak için kullanılan Microsoft Entra kullanıcı kimliğine eşlenmiş bir veritabanı kullanıcısı oluşturun.

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

  • Bağlayıcının ilgili tablolara başarıyla yazabilmesi ve bu tablolardan okunabilmesi için tabloların tanımlanacağı şemayı oluşturun.

    CREATE SCHEMA [<schema_name>];

Kimlik Doğrulaması

Microsoft Entra Id tabanlı kimlik doğrulaması

Microsoft Entra Id tabanlı kimlik doğrulaması tümleşik bir kimlik doğrulama yaklaşımıdır. Kullanıcının Azure Synapse Analytics Çalışma Alanı'nda başarıyla oturum açması gerekir.

Temel kimlik doğrulama

Kullanıcının username ve password seçeneklerini yapılandırmasını gerektiren temel bir kimlik doğrulama yaklaşımı. Azure Synapse Ayrılmış SQL Havuzu'ndaki tablolardan okuma ve tabloya yazmayla ilgili yapılandırma parametreleri hakkında bilgi edinmek için yapılandırma seçenekleri bölümüne bakın.

Yetkilendirme

Azure Data Lake Storage Nesil 2

Azure Data Lake Storage 2. Nesil erişim izinleri vermenin iki yolu vardır: Depolama Hesabı:

  • Rol tabanlı Erişim Denetimi rolü - Depolama Blob Verileri Katkıda Bulunanı rolü
    • ataması Storage Blob Data Contributor Role , Kullanıcıya Azure Depolama Blobu Kapsayıcıları'nı okuma, yazma ve silme izinleri verir.
    • RBAC, konteyner düzeyinde genel bir kontrol yaklaşımı sunar.
  • Erişim Denetim Listeleri (ACL)
    • ACL yaklaşımı, belirli bir klasör altındaki belirli yollar ve/veya dosyalar üzerinde ayrıntılı denetimler yapılmasını sağlar.
    • Kullanıcıya RBAC yaklaşımı kullanılarak zaten izinler verildiyse ACL denetimleri uygulanmaz.
    • İki geniş türde ACL izni vardır:
      • Erişim İzinleri (belirli bir düzeyde veya nesnede uygulanır).
      • Varsayılan İzinler (tüm alt nesneler oluşturulduklarında otomatik olarak uygulanır).
    • İzin türü şunlardır:
      • Execute , klasör hiyerarşileri arasında geçiş yapma veya bu hiyerarşilerde gezinme olanağı sağlar.
      • Read okuma özelliğini etkinleştirir.
      • Write yazma özelliğini etkinleştirir.
    • ACL'leri Bağlayıcının depolama konumlarından başarıyla yazabileceği ve okuyabileceği şekilde yapılandırmak önemlidir.

Not

  • Synapse Çalışma Alanı işlem hatlarını kullanarak not defterlerini çalıştırmak istiyorsanız, Synapse Çalışma Alanı varsayılan yönetilen kimliğine yukarıda listelenen erişim izinlerini de vermelisiniz. Çalışma alanının varsayılan yönetilen kimlik adı, çalışma alanının adıyla aynıdır.

  • Synapse çalışma alanını güvenli depolama hesaplarıyla kullanmak için, not defterinden yönetilen bir özel uç nokta yapılandırılmalıdır. Yönetilen özel uç noktanın, ADLS 2. Nesil depolama hesabının Private endpoint connections bölümündeki Networking bölmesinden onaylanması gerekmektedir.

Azure Synapse Ayrılmış SQL Havuzu

Azure Synapse Ayrılmış SQL Havuzu ile başarılı etkileşimi etkinleştirmek için, Ayrılmış SQL Bitiş Noktası'nda da yapılandırılmış bir Active Directory Admin kullanıcı değilseniz aşağıdaki yetkilendirme gereklidir:

  • Senaryo okuma

    • db_exporter kullanıcısına sistem saklı yordamı sp_addrolemember kullanılarak yetkilendirme yapın.

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

  • Senaryo yaz.

    • Bağlayıcı, verileri hazırlama alanından iç tablonun yönetilen konumuna yazmak için COPY komutunu kullanır.

      • Burada açıklanan gerekli izinleri yapılandırın.

      • Aşağıda aynı hızlı erişim parçası verilmiştir:

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

API belgeleri

Apache Spark için Azure Synapse Ayrılmış SQL Havuzu Bağlayıcısı - API Belgeleri.

Yapılandırma seçenekleri

Okuma veya yazma işlemini başarıyla önyüklemek ve yönetmek için Bağlayıcı belirli yapılandırma parametrelerini bekler. nesne tanımı - com.microsoft.spark.sqlanalytics.utils.Constants her parametre anahtarı için standartlaştırılmış sabitlerin listesini sağlar.

Kullanım senaryosuna göre yapılandırma seçeneklerinin listesi aşağıdadır:

  • Microsoft Entra Id tabanlı kimlik doğrulamasını kullanarak okuma
    • Kimlik bilgileri otomatik olarak eşlenir ve kullanıcının belirli yapılandırma seçeneklerini sağlaması gerekmez.
    • Azure Synapse Ayrılmış SQL Havuzu'ndaki ilgili tabloda okumak için synapsesql yönteminde üç parçalı tablo adı bağımsız değişkeni gereklidir.
  • Temel kimlik doğrulama kullanarak oku
    • Azure Synapse Ayrılmış SQL Uç Noktası
      • Constants.SERVER - Synapse Ayrılmış SQL Havuzu Uç Noktası (Server FQDN)
      • Constants.USER - SQL Kullanıcı Adı.
      • Constants.PASSWORD - SQL Kullanıcı Parolası.
    • Azure Data Lake Storage (2. Nesil) Son Nokta - Geçici Klasörler
      • Constants.DATA_SOURCE - Veri hazırlama için veri kaynağı konum parametresinde ayarlanan depolama yolu kullanılır.
  • Microsoft Entra Id tabanlı kimlik doğrulaması kullanarak yazma
    • Azure Synapse Ayrılmış SQL Uç Noktası
      • Varsayılan olarak, Bağlayıcı, yönteminin üç bölümlü tablo adı parametresinde ayarlanan synapsesql veritabanı adını kullanarak Synapse Ayrılmış SQL uç noktasını çıkarsar.
      • Alternatif olarak, kullanıcılar sql uç noktasını belirtmek için seçeneğini kullanabilir Constants.SERVER . Bitiş noktasının ilgili veritabanını ilgili şemayla barındırdığından emin olun.
    • Azure Data Lake Storage (2. Nesil) Son Nokta - Geçici Klasörler
      • İç Veri Tablosu Türü için:
        • Constants.TEMP_FOLDER veya Constants.DATA_SOURCE seçeneğini yapılandırın.
        • Kullanıcı Constants.DATA_SOURCE seçeneğini sağlamayı seçerse, hazırlama klasörü, DataSource'tan alınan location değeri kullanılarak türetilir.
        • Her ikisi de sağlanmışsa seçenek Constants.TEMP_FOLDER değeri kullanılır.
        • Hazırlama klasörü seçeneği olmadığında, Bağlayıcı çalışma zamanı yapılandırmasına göre bir klasör türetir - spark.sqlanalyticsconnector.stagingdir.prefix.
      • Dış Tablo Türü için:
        • Constants.DATA_SOURCE gerekli bir yapılandırma seçeneğidir.
        • Bağlayıcı, veri kaynağının konum parametresinde ayarlanan depolama yolunu location yöntemindeki synapsesql bağımsız değişkeniyle birlikte kullanarak, dış tablo verilerini saklamak için mutlak yolu türetir.
        • location synapsesql yöntemine bağımsız değişken belirtilmezse bağlayıcı, konum değerini <base_path>/dbName/schemaName/tableName olarak türetecektir.
  • Temel kimlik doğrulaması kullanarak yazma
    • Azure Synapse Ayrılmış SQL Uç Noktası
      • Constants.SERVER - - Synapse Ayrılmış SQL Havuzu Uç Noktası (Server FQDN).
      • Constants.USER - SQL Kullanıcı Adı.
      • Constants.PASSWORD - SQL Kullanıcı Parolası.
      • Constants.STAGING_STORAGE_ACCOUNT_KEY barındıran Constants.TEMP_FOLDERS Depolama Hesabı ile ilişkili (yalnızca iç tablo türleri) veya Constants.DATA_SOURCE.
    • Azure Data Lake Storage (2. Nesil) Son Nokta - Geçici Klasörler
      • SQL basit kimlik doğrulama bilgileri, depolama uç noktalarına erişim için geçerli değildir.
      • Bu nedenle, Azure Data Lake Storage 2. Nesil bölümünde açıklandığı gibi ilgili depolama erişim izinlerini atadığınızdan emin olun.

Kod şablonları

Bu bölümde, Apache Spark için Azure Synapse Ayrılmış SQL Havuzu Bağlayıcısı'nı kullanmayı ve çağırmayı açıklayan başvuru kodu şablonları sunulur.

Not

Python'da Bağlayıcıyı Kullanma-

Azure Synapse Ayrılmış SQL Havuzundan okuma

Okuma İsteği - synapsesql yöntem imzası

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

Microsoft Entra Id tabanlı kimlik doğrulamasını kullanarak bir tablodan okuma

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

Microsoft Entra Id tabanlı kimlik doğrulamasını kullanarak sorgudan okuma

Not

Sorgudan okurken kısıtlamalar:

  • Tablo adı ve sorgu aynı anda belirtilemez.
  • Yalnızca belirli sorgulara izin verilir. DDL ve DML SQL'lerine izin verilmez.
  • Bir sorgu belirtildiğinde veri çerçevesi üzerindeki seçme ve filtreleme seçenekleri SQL ayrılmış havuzuna gönderilmez.
  • Sorgudan okuma yalnızca Spark 3'te kullanılabilir.
//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()

Temel kimlik doğrulamasını kullanarak tablodan okuma

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

Temel kimlik doğrulamasını kullanarak sorgudan okuma

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

Azure Synapse Ayrılmış SQL Havuzuna yazma

Yazma İsteği - synapsesql yöntem imzası

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

Microsoft Entra Id tabanlı kimlik doğrulaması kullanarak yazma

Aşağıda, yazma senaryoları için Bağlayıcı'nın nasıl kullanılacağını açıklayan kapsamlı bir kod şablonu bulunur:

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

Temel kimlik doğrulaması kullanarak yazma

Aşağıdaki kod parçacığı, SQL temel kimlik doğrulaması yaklaşımını kullanarak yazma isteği göndermek için Microsoft Entra Id tabanlı kimlik doğrulaması kullanarak yazma bölümünde açıklanan yazma tanımının yerini alır:

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

Temel bir kimlik doğrulama yaklaşımında, bir kaynak depolama yolundaki verileri okumak için diğer yapılandırma seçenekleri gereklidir. Aşağıdaki kod parçacığı, Hizmet Sorumlusu kimlik bilgilerini kullanarak bir Azure Data Lake Storage 2. Nesil veri kaynağından okuma örneği sağlar:

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

Desteklenen DataFrame kaydetme modları

Azure Synapse Ayrılmış SQL Havuzu'nda hedef tabloya kaynak veriler yazılırken aşağıdaki kaydetme modları desteklenir:

  • ErrorIfExists (varsayılan kaydetme modu)
    • Hedef tablo varsa, çağrıya döndürülen bir özel durumla yazma işlemi iptal edilir. Aksi halde, hazırlama klasörlerinden alınan verilerle yeni bir tablo oluşturulur.
  • Yok saymak
    • Hedef tablo mevcutsa, yazma işlemi, bir hata döndürmeden yazma talebini yoksayar. Aksi halde, hazırlama klasörlerinden alınan verilerle yeni bir tablo oluşturulur.
  • Üzerine yaz
    • Hedef tablo varsa, hedefteki mevcut veriler hazırlama klasörlerindeki verilerle değiştirilir. Aksi halde, hazırlama klasörlerinden alınan verilerle yeni bir tablo oluşturulur.
  • Ekleme
    • Hedef tablo varsa, yeni veriler bu tabloya eklenir. Aksi halde, hazırlama klasörlerinden alınan verilerle yeni bir tablo oluşturulur.

Yazma isteği geri çağırma işleyicisi

Yeni yazma yolu API'sinde yapılan değişiklikler, istemciye yazma sonrası ölçümlerin anahtar-değer> eşlemesini sağlayan deneysel bir özellik sağladı. Ölçümlerin anahtarları yeni Nesne tanımında tanımlanır: Constants.FeedbackConstants. Ölçümler, geri çağırma tutamacını (a Scala Function) geçirerek JSON dizesi olarak alınabilir. İşlev imzası aşağıdadır:

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

Aşağıda bazı önemli ölçümler verilmiştir (deve durumunda sunulmuştur):

  • WriteFailureCause
  • DataStagingSparkJobDurationInMilliseconds
  • NumberOfRecordsStagedForSQLCommit
  • SQLStatementExecutionDurationInMilliseconds
  • rows_processed

Aşağıda, yazma sonrası ölçümlere sahip örnek bir JSON dizesi verilmiştir:

{
 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
}

Diğer kod örnekleri

Hücreler arasında malzeme verileri kullanma

Spark DataFrame'ler createOrReplaceTempView , geçici bir görünüm kaydederek başka bir hücrede getirilen verilere erişmek için kullanılabilir.

  • Verilerin alındığı hücre (örneğin, Not Defteri dil tercihi olarak 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>")
  • Şimdi Not Defteri'nde PySpark (Python) dil tercihini olarak değiştirin ve kayıtlı görünümden veri getirin <temporary_view_name>
        spark.sql("select * from <temporary_view_name>").show()

Yanıt işleme

Çağırmanın synapsesql iki olası son durumu vardır: Başarılı veya Başarısız Durum. Bu bölümde, her senaryo için istek yanıtını işleme açıklanmaktadır.

İstek yanıtını oku

Tamamlandıktan sonra, hücrenin çıkışında okunan yanıt parçası görüntülenir. Geçerli hücrede bir hata olması, sonraki hücre çalıştırmalarını da iptal eder. Ayrıntılı hata bilgileri Spark Uygulama Günlükleri'nde bulunur.

İstek yanıtı yazma

Varsayılan olarak, hücre çıkışına bir yazma yanıtı yazdırılır. Hata durumunda geçerli hücre başarısız olarak işaretlenir ve sonraki hücre yürütmeleri iptal edilir. Diğer yaklaşım ise geri çağırma tutamacı seçeneğini synapsesql yöntemine geçirmektir. Geri çağırma tutamacı, yazma yanıtına programatik erişim sağlar.

Dikkat edilecek diğer noktalar

  • Azure Synapse Ayrılmış SQL Havuzu tablolarından okurken:
    • Bağlayıcının sütun ayıklama özelliğinden yararlanmak için DataFrame'e gerekli filtreleri uygulamayı göz önünde bulundurun.
    • Okuma senaryosu TOP(n-rows) yan tümcesini, SELECT sorgu cümlelerini çerçevelerken desteklemez. Verileri sınırlama seçeneği DataFrame'in limit(.) yan tümcesini kullanmaktır.
  • Azure Synapse Ayrılmış SQL Havuzu tablolarına yazarken:
    • İç tablo türleri için:
      • Tablolar ROUND_ROBIN veri dağıtımıyla oluşturulur.
      • Sütun türleri, verinin kaynaktan okunduğu DataFrame'den belirlenir. Dize sütunları NVARCHAR(4000) ile eşlenir.
    • Dış tablo türleri için:
      • DataFrame'in ilk paralelliği, dış tablo için veri organizasyonunu yönlendirir.
      • Sütun türleri, verinin kaynaktan okunduğu DataFrame'den belirlenir.
    • spark.sql.files.maxPartitionBytes ve DataFrame repartition parametresi ayarlanarak yürütücüler arasında daha iyi veri dağıtımı elde edilebilir.
    • Büyük veri kümeleri yazarken, işlem boyutunu sınırlayan DWU Performans Düzeyi ayarının etkisini dikkate almak önemlidir.
  • Okuma ve yazma performansını etkileyebilecek azaltma davranışlarını saptamak için Azure Data Lake Storage 2. Nesil kullanım eğilimlerini izleyin.

İlgili içerik

  • Last updated on