使用适用于 SQL Server 和 Azure SQL 的 Apache Spark 连接器

项目
2019-11-04

重要

Microsoft SQL Server 2019 大数据群集附加产品将停用。对 SQL Server 2019 大数据群集的支持将于 2025 年 2 月 28 日结束。具有软件保障的 SQL Server 2019 的所有现有用户都将在平台上获得完全支持，在此之前，该软件将继续通过 SQL Server 累积更新进行维护。有关详细信息，请参阅公告博客文章和 Microsoft SQL Server 平台上的大数据选项。

适用于 SQL Server 和 Azure SQL 的 Apache Spark 连接器是一种高性能连接器，可用于在大数据分析中使用事务数据，并为即席查询或报告保留结果。连接器允许将任何 SQL 数据库（本地或云中）用作 Spark 作业的输入数据源或输出数据接收器。连接器使用 SQL Server 批量写入 API。任何批量写入参数都可以由用户作为可选参数进行传递，并由连接器通过 as-is 传递到基础 API。有关批量写入操作的详细信息，请参阅将批量复制与JDBC驱动程序配合使用。

默认情况下，连接器包含在 SQL Server 大数据群集中。

在开源存储库中了解有关连接器的详细信息。有关示例，请参阅示例。

写入新的 SQL 表

谨慎

在 overwrite 模式下，连接器首先删除该表（如果表已存在于数据库中）。默认情况下。请谨慎使用此选项以避免意外数据丢失。

如果使用overwrite模式，但不使用选项truncate，则重新创建表时，索引将丢失。例如，列存储表将成为堆。如果要维护现有索引，请指定选项truncate并给予值true。例如 .option("truncate",true)

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 模式的连接器示例。在运行脚本之前，请将 ? 替换为您帐户的值。

# 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 模式安全性中，在用户生成密钥选项卡文件后，用户需要在连接器实例化期间提供 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()

后续步骤

有关大数据群集的详细信息，请参阅如何在 Kubernetes 部署 SQL Server 大数据群集

通过