Conector do Spark para bancos de dados SQL

O conector Spark para bancos de dados SQL é uma biblioteca de alto desempenho que permite ler e gravar em bancos de dados SQL Server, SQL do Azure e bancos de dados SQL no Fabric. O conector oferece os seguintes recursos:

• Use o Spark para executar grandes operações de escrita e leitura no Banco de Dados SQL do Azure, Instância Gerenciada de SQL do Azure, SQL Server no Azure VM e bancos de dados SQL no Fabric.
• Quando você usa uma tabela ou uma exibição, o conector dá suporte a modelos de segurança definidos no nível do mecanismo SQL. Esses modelos incluem segurança em nível de objeto (OLS), segurança em nível de linha (RLS) e segurança em nível de coluna (CLS).

O conector é pré-instalado no runtime do Fabric, portanto, você não precisa instalá-lo separadamente.

Authentication

A autenticação Microsoft Entra é integrada ao Fabric.

• Quando você entra no workspace do Fabric, suas credenciais são passadas automaticamente para o mecanismo SQL para autenticação e autorização.
• Exige que a ID do Microsoft Entra seja habilitada e configurada em seu mecanismo de banco de dados SQL.
• Nenhuma configuração extra será necessária no código do Spark se a ID do Microsoft Entra estiver configurada. As credenciais são mapeadas automaticamente.

Você também pode usar o método de autenticação SQL (especificando um nome de usuário e senha do SQL) ou um principal de serviço (fornecendo um token de acesso do Azure para autenticação baseada em aplicativo).

Permissions

Para usar o conector do Spark, sua identidade, seja um usuário ou um aplicativo, deve ter as permissões de banco de dados necessárias para o mecanismo SQL de destino. Essas permissões são necessárias para ler ou gravar em tabelas e exibições.

Para o Banco de Dados SQL do Azure, a Instância Gerenciada de SQL do Azure e o SQL Server na VM do Azure:

• A identidade que executa a operação normalmente precisa db_datawriter e db_datareader permissões e, opcionalmente db_owner , para controle total.

Para um banco de dados SQL no Fabric:

• A identidade normalmente precisa das permissões db_datawriter e db_datareader, e, opcionalmente, db_owner.
• A identidade também precisa de, no mínimo, permissão de leitura no banco de dados SQL no Fabric, no nível do item.

Exemplos de uso e código

Nesta seção, fornecemos exemplos de código para demonstrar como usar o conector Spark para bancos de dados SQL com eficiência. Esses exemplos abrangem vários cenários, incluindo ler e gravar em tabelas SQL e configurar as opções do conector.

Opções com suporte

A opção mínima necessária é url como "jdbc:sqlserver://<server>:<port>;database=<database>;" ou definir spark.mssql.connector.default.url.

• Quando o url for fornecido:

  • Sempre use url como primeira preferência.
  • Se spark.mssql.connector.default.url não estiver definido, o conector o definirá e o reutilizará para uso futuro.

• Quando o url não for fornecido:

  • Se spark.mssql.connector.default.url estiver definido, o conector usará o valor da configuração do spark.
  • Se spark.mssql.connector.default.url não estiver definido, um erro será gerado porque os detalhes necessários não estão disponíveis.

Esse conector dá suporte às opções definidas aqui: Opções JDBC do SQL DataSource

O conector também dá suporte às seguintes opções:

Outras opções de API em massa podem ser configuradas e DataFrame são passadas para as APIs de cópia em massa durante a gravação.

Exemplo de gravação e leitura

O código a seguir utiliza autenticação automática do Microsoft Entra ID para demonstrar essas operações:

• Escreva um DataFrame: df.write.option("...", "...").mssql("<schema>.<table>").
• Leia uma tabela: spark.read.option("...", "...").mssql("<schema>.<table>").
• Execute uma consulta personalizada: spark.read.option("...", "...").option("query", "<your-custom-query>").mssql().

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)
df.write.mode("overwrite").option("url", url).mssql("dbo.publicExample")
spark.read.option("url", url).mssql("dbo.publicExample").show()
spark.read.option("url", url).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

url = "jdbc:sqlserver://<server>:<port>;database=<database2>;" # different database
df.write.mode("overwrite").option("url", url).mssql("dbo.tableInDatabase2") # default url is updated
spark.read.mssql("dbo.tableInDatabase2").show() # no url option specified and will use database2

import com.microsoft.sqlserver.jdbc.spark.SparkSqlImplicits._
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
val url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
val row_data = Seq(
  Row("Alice", 1),
  Row("Bob", 2),
  Row("Charlie", 3)
)
val schema = List(
  StructField("Name", StringType, nullable = true),
  StructField("Age", IntegerType, nullable = true)
)
val df = spark.createDataFrame(spark.sparkContext.parallelize(row_data), StructType(schema))
df.write.mode("overwrite").option("url", url).mssql("dbo.publicExample")
spark.read.option("url", url).mssql("dbo.publicExample").show
spark.read.option("url", url).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show // Read with a custom query

val url = "jdbc:sqlserver://<server>:<port>;database=<database2>;" // different database
df.write.mode("overwrite").option("url", url).mssql("dbo.tableInDatabase2") // default url is updated
spark.read.mssql("dbo.tableInDatabase2").show // no url option specified and will use database2

Você também pode selecionar colunas, aplicar filtros e usar outras opções ao ler dados do mecanismo de banco de dados SQL.

Exemplos de autenticação

Os exemplos a seguir mostram como usar métodos de autenticação além do Microsoft Entra ID, como service principal (token de acesso) e a autenticação SQL.

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)

from azure.identity import ClientSecretCredential
credential = ClientSecretCredential(tenant_id="", client_id="", client_secret="") # service principal app
scope = "https://database.windows.net/.default"
token = credential.get_token(scope).token

df.write.mode("overwrite").option("url", url).option("accesstoken", token).mssql("dbo.publicExample")
spark.read.option("accesstoken", token).mssql("dbo.publicExample").show()
spark.read.option("accesstoken", token).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)
df.write.mode("overwrite").option("url", url).option("user", "").option("password", "").mssql("dbo.publicExample")
spark.read.option("user", "").option("password", "").mssql("dbo.publicExample").show()
spark.read.option("user", "").option("password", "").option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

Modos de salvamento de DataFrame com suporte

Ao gravar dados do Spark em bancos de dados SQL, você pode escolher entre vários modos de salvamento. Os modos de salvamento controlam como os dados são gravados quando a tabela de destino já existe e podem afetar o esquema, os dados e a indexação. Entender esses modos ajuda você a evitar perdas ou alterações inesperadas de dados.

Esse conector dá suporte às opções definidas aqui: Funções de salvamento do Spark

• ErrorIfExists (modo de salvamento padrão): se a tabela de destino existir, a gravação será anulada e uma exceção será retornada. Caso contrário, uma nova tabela será criada com dados.

• Ignorar: se a tabela de destino existir, a gravação ignorará a solicitação e não retornará um erro. Caso contrário, uma nova tabela será criada com dados.

• Substituir: Caso a tabela de destino já exista, ela será descartada, recriada e novos dados serão adicionados.

  Observação

  Quando você usa overwrite, perde o esquema original da tabela (especialmente os tipos de dados exclusivos do MSSQL) e os índices da tabela. O esquema é substituído pelo esquema inferido do seu DataFrame Spark. Para evitar perder o esquema e os índices, adicione .option("truncate", true).

• Acrescentar: se a tabela de destino existir, novos dados serão acrescentados a ela. Caso contrário, uma nova tabela será criada com dados.

Troubleshoot

Quando o processo é concluído, a saída da operação de leitura do Spark é exibida na área de saída da célula. Erros de com.microsoft.sqlserver.jdbc.SQLServerException vêm diretamente do SQL Server. Você pode encontrar informações detalhadas de erro nos logs de aplicativos do Spark.

Escritas em massa exigem que dados recebidos se ajustem aos tipos de dados SQL alvo. Se os dados não se encaixarem, você pode receber este erro:

Caused by: com.microsoft.sqlserver.jdbc.SQLServerException: The service has encountered an error processing your request. Please try again. Error code 4815.

Por exemplo, esse erro pode ocorrer quando a tabela SQL de destino usa datetime, mas um valor TimestampType de Spark recebido, como 2025-01-01 10:30:00.123456, tem mais precisão do que datetime suporta.

Para resolver o erro, use uma destas abordagens:

• Converta, trunque ou transforme os dados recebidos para adequá-los ao tipo de dados SQL de destino. Por exemplo, trunque o valor para três dígitos de precisão de segundos fracionários: 2025-01-01 10:30:00.123000.
• Permita que o conector recrie o esquema da tabela definindo .option("truncate", false). O conector mapeia o tipo de timestamp do Spark para SQL datetime2.

Conteúdo relacionado

• Banco de Dados SQL do Azure
• Banco de dados SQL no Fabric
• Banco de Dados SQL do Azure – Autenticação e autorização