Использование соединителя Apache Spark для SQL Server и SQL Azure

Соединитель Apache Spark для SQL Server и Azure SQL — это высокопроизводительный соединитель, который позволяет использовать транзакционные данные в аналитике больших данных и сохранять результаты для нерегламентированных запросов или отчетов. Соединитель позволяет использовать любую базу данных SQL, локальную или облачную, в качестве источника входных данных или приемника выходных данных для заданий Spark. Соединитель использует API-интерфейсы массовой записи SQL Server. Любые параметры массовой записи могут передаваться пользователем в качестве необязательного параметра и передаются соединителем в базовый API. Дополнительные сведения об операциях массовой записи см. в разделе Использование массового копирования с помощью JDBC Driver.

Этот соединитель включается по умолчанию в состав кластеры SQL Server больших данных.

Дополнительные сведения о соединителе см. в репозитории с открытым исходным кодом. Примеры см. в образцах.

Запись в новую таблицу SQL

server_name = "jdbc:sqlserver://{SERVER_ADDR}"
database_name = "database_name"
url = server_name + ";" + "databaseName=" + database_name + ";"

table_name = "table_name"
username = "username"
password = "password123!#" # Please specify password here

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("overwrite") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

Добавление в таблицу SQL

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("append") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

Указание уровня изоляции

По умолчанию этот соединитель использует уровень изоляции READ_COMMITTED при выполнении массовой вставки в базу данных. Если вы хотите переопределить его на другой уровень изоляции, используйте параметр mssqlIsolationLevel, как показано ниже.

    .option("mssqlIsolationLevel", "READ_UNCOMMITTED") \

Чтение из таблицы SQL

jdbcDF = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("user", username) \
        .option("password", password).load()

Режим без Active Directory

В режиме без Active Directory каждый пользователь имеет имя пользователя и пароль, которые необходимо предоставить в качестве параметров во время создания экземпляра соединителя для выполнения операций чтения и записи.

Ниже приведен пример создания экземпляра соединителя для режима без Active Directory. Перед выполнением скрипта замените ? значением для своей учетной записи.

# Note: '?' is a placeholder for a necessary user-specified value
connector_type = "com.microsoft.sqlserver.jdbc.spark" 

url = "jdbc:sqlserver://master-p-svc;databaseName=?;"
writer = df.write \ 
   .format(connector_type)\ 
   .mode("overwrite") 
   .option("url", url) \ 
   .option("user", ?) \ 
   .option("password",?) 
writer.save() 

Режим Active Directory

В режиме безопасности с Active Directory после того, как пользователь создал файл keytab, он должен указать principal и keytab в качестве параметров во время создания экземпляра соединителя.

В этом режиме драйвер загружает файл keytab в соответствующие контейнеры исполнителя. Затем исполнители используют имя участника и keytab для создания маркера, который используется для создания соединителя JDBC для чтения и записи.

Ниже приведен пример создания экземпляра соединителя для режима с Active Directory. Перед выполнением скрипта замените ? значением для своей учетной записи.

# Note: '?' is a placeholder for a necessary user-specified value
connector_type = "com.microsoft.sqlserver.jdbc.spark"

url = "jdbc:sqlserver://master-p-svc;databaseName=?;integratedSecurity=true;authenticationScheme=JavaKerberos;" 
writer = df.write \ 
   .format(connector_type)\ 
   .mode("overwrite") 
   .option("url", url) \ 
   .option("principal", ?) \ 
   .option("keytab", ?)   

writer.save() 

Дальнейшие действия

Дополнительные сведения о кластерах больших данных см. в статье Развертывание Кластеры больших данных SQL Server в Kubernetes.

У вас есть отзывы или рекомендации по функциям для кластеров больших данных SQL Server? Напишите нам в разделе "Отзывы от кластерах больших данных SQL Server".