Поделиться через

Соединитель выделенного пула SQL в Azure Synapse для Apache Spark

  • Статья

Введение

Соединитель выделенного пула SQL в Azure Synapse для Apache Spark в Azure Synapse Analytics эффективно передает большие наборы данных между средой выполнения Apache Spark и выделенным пулом SQL. Соединитель поставляется в виде библиотеки по умолчанию с рабочей областью Azure Synapse. Соединитель реализован на языке Scala. Соединитель поддерживает Scala и Python. Чтобы использовать коннектор с другими вариантами языка записных книжек, используйте магическую команду Spark %%spark.

Соединитель предоставляет на высоком уровне такие возможности:

  • Чтение из выделенного пула SQL Azure Synapse:
    • Чтение больших наборов данных из выделенных таблиц пула SQL Synapse (внутренних и внешних) и представлений.
    • Комплексная поддержка отправки предикатов, где фильтры в кадре данных сопоставляются с соответствующим отправленным предикатом SQL.
    • Поддержка оптимизации столбцов.
    • Поддержка передачи обработки запроса на нижние уровни.
  • Запись в выделенный SQL-пул Azure Synapse
    • Загрузка данных большого объема во внутренние и внешние типы таблиц.
    • Поддерживает следующие предпочтения режима сохранения DataFrame:
      • Append
      • ErrorIfExists
      • Ignore
      • Overwrite
    • Функция записи в тип внешней таблицы поддерживает форматы Parquet и текстовые файлы с разделителями (пример — CSV).
    • Для записи данных во внутренние таблицы коннектор теперь использует инструкцию COPY вместо подхода CETAS/CTAS.
    • Улучшения для оптимизации производительности сквозной пропускной способности записи.
    • Представляет дополнительный обработчик обратного вызова (аргумент функции Scala), который клиенты могут использовать для получения метрик после записи
      • Ниже приведены несколько примеров: количество записей, длительность выполнения определенного действия и причина сбоя.

Подход к оркестрации

Читать

Высокоуровневая схема потока данных, которая описывает оркестрацию соединителя при запросе на чтение.

Написать

Высокоуровневая схема потока данных, которая описывает оркестрацию соединителя при запросе на запись.

Предварительные условия

Предварительные требования, такие как настройка необходимых ресурсов Azure и действия по их настройке, рассматриваются в этом разделе.

Ресурсы Azure

Проверьте и настройте следующие зависимые ресурсы Azure:

Подготовка базы данных

Подключитесь к базе данных выделенного пула SQL в Synapse и выполните приведенные ниже инструкции по настройке:

  • Создайте пользователя базы данных, сопоставленного с удостоверением пользователя Microsoft Entra, используемого для входа в рабочую область Azure Synapse.

    CREATE USER [[email protected]] FROM EXTERNAL PROVIDER;

  • Создайте схему, в которой будут определены таблицы, чтобы соединитель мог успешно выполнять запись и чтение в соответствующих таблицах.

    CREATE SCHEMA [<schema_name>];

Проверка подлинности

Проверка подлинности на основе идентификатора Microsoft Entra

Проверка подлинности на основе идентификатора Microsoft Entra — это интегрированный подход к проверке подлинности. Пользователь должен успешно войти в рабочую область Azure Synapse Analytics.

Обычная проверка подлинности

Для обычной проверки подлинности пользователь должен настроить параметры username и password. Дополнительные сведения о параметрах конфигурации см. в разделе Параметры конфигурации для чтения и записи в таблицы в выделенном пуле SQL в Azure Synapse.

Авторизация

Azure Data Lake Storage 2-го поколения

Предоставить разрешения на доступ к учетной записи службы хранилища Azure Data Lake Storage 2-го поколения можно двумя способами:

  • Роль управления доступом на основе ролей — роль "Управляющий данными BLOB-хранилища"
    • Назначение Storage Blob Data Contributor Role предоставляет пользователю разрешения на чтение, запись и удаление данных из контейнеров Blob в хранилище Azure.
    • RBAC обеспечивает грубый контроль на уровне контейнера.
  • Списки управления доступом (ACL)
    • Метод ACL позволяет точно контролировать доступ к определенным путям и файлам в конкретной папке.
    • Проверки ACL не применяются, если пользователю уже предоставлены разрешения с применением RBAC.
    • Существует два широких типа разрешений ACL:
      • Разрешения на доступ (применяются на определенном уровне или объекте).
      • Разрешения по умолчанию (автоматически применяются для всех дочерних объектов в момент их создания).
    • К типу разрешений относятся следующие:
      • Execute позволяет просматривать иерархии папок и перемещаться по ним.
      • Read обеспечивает возможность чтения.
      • Write обеспечивает возможность записи.
    • Необходимо настроить ACL так, чтобы соединитель мог успешно выполнять запись и чтение из хранилища.

Примечание.

  • Если вы хотите запускать записные книжки с помощью конвейеров рабочей области Synapse, необходимо также предоставить указанные выше разрешения на доступ к управляемому удостоверению рабочей области Synapse по умолчанию. Имя управляемого удостоверения рабочей области по умолчанию такое же, как имя рабочей области.

  • Чтобы использовать рабочую область Synapse с защищенными учетными записями хранения, необходимо настроить управляемую частную конечную точку из записной книжки. Управляемая частная конечная точка должна быть утверждена в разделе Private endpoint connections учетной записи хранения ADLS 2-го поколения на панели Networking.

Выделенный пул SQL в Azure Synapse

Чтобы обеспечить успешное взаимодействие с выделенным пулом SQL в Azure Synapse, необходимо выполнить следующую авторизацию, за исключением случая когда вы являетесь пользователем, так же настроенным как Active Directory Admin на выделенной конечной точке SQL:

  • Чтение сценария

    • Предоставьте пользователю db_exporter доступ к системной хранимой процедуре sp_addrolemember.

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

  • Написать сценарий

    • Соединитель использует команду COPY для записи данных из промежуточного расположения в управляемое расположение внутренней таблицы.

      • Настройте необходимые разрешения, как описано здесь.

      • Ниже приведен соответствующий фрагмент кода для быстрого доступа:

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

Соединитель выделенного пула SQL в Azure Synapse для Apache Spark — документация по API.

Варианты конфигурации

Для успешной начальной загрузки и оркестрации операции чтения или записи соединитель ожидает определенные параметры конфигурации. Определение объекта com.microsoft.spark.sqlanalytics.utils.Constants предоставляет список стандартизированных констант для каждого ключа параметра.

Ниже приведен список параметров конфигурации на основе сценария использования:

  • Чтение с использованием аутентификации по идентификатору Microsoft Entra
    • Учетные данные автоматически сопоставляются, и пользователю не требуется предоставлять определенные параметры конфигурации.
    • Для чтения из соответствующей таблицы в выделенном пуле SQL Azure Synapse требуется аргумент с именем таблицы, состоящим из трех частей, в методе synapsesql.
  • Чтение с помощью обычной проверки подлинности
    • Выделенная конечная точка SQL Azure Synapse
      • Constants.SERVER: конечная точка выделенного пула SQL в Synapse (полное доменное имя сервера)
      • Constants.USER: имя пользователя SQL.
      • Constants.PASSWORD: пароль пользователя SQL.
    • Конечная точка Azure Data Lake Storage (2-го поколения) — промежуточные папки
      • Constants.DATA_SOURCE: путь хранения, заданный в параметре расположения источника данных, используется для промежуточного хранения данных.
  • Выполнение записи с аутентификацией через Microsoft Entra ID
    • Выделенная конечная точка SQL Azure Synapse
      • По умолчанию соединитель определяет конечную точку выделенного SQL Synapse, используя имя базы данных, установленное в параметре трехчастного имени таблицы метода synapsesql.
      • Кроме того, пользователи могут использовать параметр Constants.SERVER для указания конечной точки SQL. Убедитесь, что конечная точка размещает соответствующую базу данных с соответствующей схемой.
    • Конечная точка Azure Data Lake Storage (2-го поколения) — промежуточные папки
      • Для внутреннего типа таблицы:
        • Настройте параметр Constants.TEMP_FOLDER или Constants.DATA_SOURCE.
        • Если пользователь выбрал предоставить опцию Constants.DATA_SOURCE, промежуточная папка будет определена по значению location в источнике данных.
        • Если указаны оба параметра, используется значение параметра Constants.TEMP_FOLDER.
        • При отсутствии опции промежуточной папки соединитель определяет её на основе заданной конфигурации среды выполнения spark.sqlanalyticsconnector.stagingdir.prefix.
      • Для внешнего типа таблицы:
        • Constants.DATA_SOURCE - обязательный параметр конфигурации.
        • Соединитель использует путь к хранилищу, который задaн в параметре расположения источника данных, в сочетании с аргументом location метода synapsesql и выводит абсолютный путь для сохранения данных внешней таблицы.
        • Если аргумент location метода synapsesql не указан, соединитель выводит значение расположения как <base_path>/dbName/schemaName/tableName.
  • Запись с использованием обычной проверки подлинности
    • Выделенная конечная точка SQL Azure Synapse
      • Constants.SERVER: конечная точка выделенного пула SQL Synapse (полное доменное имя сервера).
      • Constants.USER: имя пользователя SQL.
      • Constants.PASSWORD: пароль пользователя SQL.
      • Constants.STAGING_STORAGE_ACCOUNT_KEY, связанный с учетной записью в хранилище, где размещаются Constants.TEMP_FOLDERS (только внутренние типы таблиц) или Constants.DATA_SOURCE.
    • Конечная точка Azure Data Lake Storage (2-го поколения) — промежуточные папки
      • Учетные данные SQL для базовой аутентификации не применяются к точкам доступа в хранилище.
      • Поэтому назначьте соответствующие разрешения на доступ к хранилищу, как описано в разделе Azure Data Lake Storage 2-го поколения.

Шаблоны кода

В этом разделе представлены справочные шаблоны кода и описано, как использовать и вызывать соединитель выделенного пула SQL в Azure Synapse для Apache Spark.

Примечание.

Использование соединителя в Python

Чтение данных из выделенного пула SQL в Azure Synapse

Запрос на чтение — сигнатура метода synapsesql

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

Чтение из таблицы с помощью проверки подлинности на основе идентификатора 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()

Чтение из запроса с помощью аутентификации на основе идентификатора Microsoft Entra ID

Примечание.

Ограничения при чтении из запроса:

  • Имя таблицы и запрос нельзя указать одновременно.
  • Разрешены только определённые запросы. Не допускаются DDL и DML SQLs.
  • Операции выборки и фильтрации в dataframe не передаются в выделенный пул SQL при указании запроса.
  • Чтение из запроса доступно только в 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()

Чтение данных из таблицы с помощью базовой аутентификации

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

Чтение из запроса с помощью базовой аутентификации

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

Запись данных в выделенный пул SQL в Azure Synapse

Запрос на запись — сигнатура метода synapsesql

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

Запись с использованием аутентификации на основе Microsoft Entra ID.

В следующем комплексном шаблоне кода описывается использование соединителя для сценариев записи:

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

Пишите, используя базовую аутентификацию

Следующий фрагмент кода заменяет определение записи, описанное в разделе "Запись с использованием проверки подлинности на основе идентификатора Microsoft Entra", чтобы отправить запрос на запись с использованием 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))

При обычной проверке подлинности, чтобы считывать данные из исходного пути к хранилищу, требуются другие параметры конфигурации. В следующем фрагменте кода приведен пример для чтения данных из Azure Data Lake Storage второго поколения с помощью учетных данных служебного принципала.

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

Поддерживаемые режимы сохранения DataFrame

При записи исходных данных в целевую таблицу в выделенном пуле SQL в Azure Synapse поддерживаются следующие режимы сохранения:

  • ErrorIfExists (режим сохранения по умолчанию)
    • Если целевая таблица уже существует, то запись прерывается с исключением, возвращаемым вызывающему. В противном случае создается новая таблица с данными из промежуточных папок.
  • Игнорировать
    • Если целевая таблица существует, операция записи игнорирует запрос на запись и не возвращает ошибку. В противном случае создается новая таблица с данными из промежуточных папок.
  • Перезаписать
    • Если целевая таблица существует, существующие данные в месте назначения заменяются данными из промежуточных папок. В противном случае создается новая таблица с данными из промежуточных папок.
  • Добавлять
    • Если целевая таблица существует, к ней добавляются новые данные. В противном случае создается новая таблица с данными из промежуточных папок.

Дескриптор обратного вызова для запроса записи

В новом API маршрута записи представлена экспериментальная функция, которая предоставляет клиенту отображение ключей и значений для метрик после записи. Ключи для метрик определяются в новом определении объекта Constants.FeedbackConstants. Метрики можно получить в виде строки JSON, передав дескриптор обратного вызова (Scala Function). Следующий фрагмент кода является сигнатурой функции:

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

Ниже приведены некоторые важные метрики (в "верблюжьем" стиле):

  • WriteFailureCause
  • DataStagingSparkJobDurationInMilliseconds
  • NumberOfRecordsStagedForSQLCommit
  • SQLStatementExecutionDurationInMilliseconds
  • rows_processed

Ниже приведен пример строки JSON с метриками после записи:

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

Дополнительные примеры кода

Использование материализованных данных между ячейками

createOrReplaceTempView DataFrame Spark можно использовать для доступа к данным, извлекаемым в другой ячейке, регистрируя временное представление.

  • Ячейка, из которой извлекаются данные (например, с предпочтением языка записной книжки 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>")
  • Теперь измените языковые параметры записной книжки на PySpark (Python) и извлеките данные из зарегистрированного представления <temporary_view_name>
        spark.sql("select * from <temporary_view_name>").show()

Обработка ответов

Вызов synapsesql имеет два возможных конечных состояния: "Успешно" или "Сбой". В этом разделе рассматривается обработка ответа на запрос для каждого сценария.

Чтение ответа на запрос

По завершении фрагмент чтения ответа отображается в выходных данных ячейки. Сбой в текущей ячейке также отменяет последующие выполнения других ячеек. Подробные сведения об ошибке доступны в журналах приложений Spark.

Ответ на запрос о записи

По умолчанию ответ на запись выводится в выходные данные ячейки. При сбое текущая ячейка помечается как неудачная, и последующие выполнения ячеек будут прерваны. Другой подход — передача параметра дескриптора обратного вызова методу synapsesql. Дескриптор обратного вызова предоставляет программный доступ к ответу на запись.

Другие вопросы

  • При чтении из таблиц выделенного пула SQL Azure Synapse:
    • Рассмотрите возможность применения необходимых фильтров к DataFrame, чтобы воспользоваться преимуществами функции обрезки столбцов соединителя.
    • Сценарий чтения не поддерживает предложение TOP(n-rows) при обрамлении операторов запроса SELECT. Для ограничения данных используйте предложение dataFrame limit(.).
  • При записи в таблицы выделенного пула SQL Azure Synapse:
    • Для внутренних типов таблиц:
      • Таблицы создаются с распределением данных ROUND_ROBIN.
      • Типы столбцов выводятся из DataFrame, который будет считывать данные из источника. Строковые столбцы сопоставляются с NVARCHAR(4000).
    • Для внешних типов таблиц:
      • Начальный параллелизм DataFrame управляет организацией данных для внешней таблицы.
      • Типы столбцов выводятся из DataFrame, который будет считывать данные из источника.
    • Лучшего распределения данных между исполнителями можно добиться, настроив spark.sql.files.maxPartitionBytes и параметр DataFrame repartition.
    • При записи больших наборов данных важно учитывать влияние параметра уровня производительности DWU, ограничивающего размер транзакции.
  • Отслеживайте тенденции использования Azure Data Lake Storage 2-го поколения, чтобы определить поведение регулирования, которое может повлиять на производительность чтения и записи.

Связанный контент