Conector Spark para bancos de dados SQL

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

  • Use o Spark para executar grandes operações de escrita e leitura no Base de Dados SQL do Azure, Azure SQL Managed Instance, SQL Server no Azure VM e bases de dados SQL no Fabric.
  • Quando usa uma tabela ou uma vista, o conector suporta modelos de segurança definidos ao nível do motor 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 está pré-instalado no runtime do Fabric, por isso não precisas de o instalar separadamente.

Authentication

A autenticação Microsoft Entra está integrada com o Fabric.

  • Quando inicia sessão no espaço de trabalho do Fabric, as suas credenciais são automaticamente passadas para o motor SQL para autenticação e autorização.
  • Requer que o Microsoft Entra ID esteja ativado e configurado no seu motor de base de dados SQL.
  • Não é necessária nenhuma configuração extra no seu código Spark se o Microsoft Entra ID estiver configurado. As credenciais são automaticamente mapeadas.

Também pode usar o método de autenticação SQL (especificando um nome de utilizador e palavra-passe SQL) ou um principal de serviço (fornecendo um token de acesso Azure para autenticação baseada em aplicações).

Permissions

Para usar o conector Spark, a sua identidade — seja um utilizador ou uma aplicação — deve ter as permissões necessárias para a base de dados do motor SQL alvo. Estas permissões são necessárias para ler ou escrever em tabelas e vistas.

Para Base de Dados SQL do Azure, Azure SQL Managed Instance e SQL Server no Azure VM:

  • A identidade que executa a operação normalmente necessita das permissões db_datawriter e db_datareader, e opcionalmente de db_owner para controlo total.

Para uma base de dados SQL em Fabric:

  • A identidade normalmente requer permissões de db_datawriter e db_datareader, e opcionalmente db_owner.
  • A identidade também precisa de, pelo menos, permissão de leitura na base de dados SQL no Fabric ao nível do item.

Observação

Se usar um principal de serviço, este pode ser executado como uma aplicação (sem contexto de utilizador) ou como um utilizador se a personificação de utilizador estiver ativada. O principal do serviço deve ter as permissões necessárias para a base de dados necessária para as operações que pretende realizar.

Utilização e exemplos de código

Nesta secção, fornecemos exemplos de código para demonstrar como usar eficazmente o conector Spark para bases de dados SQL. Estes exemplos abrangem vários cenários, incluindo ler e escrever em tabelas SQL, e configurar as opções do conector.

Observação

Antes de uma escrita em massa, todos os dados recebidos do Spark devem encaixar nos tipos de dados SQL de destino. Ao sobrescrever ou criar uma tabela, o conector mapeia os valores Spark TimestampType e TimestampNTZType para SQL datetime2 em vez de datetime. Os tipos de carimbo temporal do Spark suportam até seis dígitos de precisão fracionária de segundo, mas o SQL datetime suporta três, o que pode causar uma incompatibilidade.

Opções suportadas

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 é fornecido:

    • Usa url sempre como primeira preferência.
    • Se spark.mssql.connector.default.url não estiver definido, o conector irá configurá-lo e reutilizá-lo para usos futuros.

  • Quando não é url fornecido:

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

Este conector suporta as opções definidas aqui: SQL DataSource JDBC Options

O conector também suporta as seguintes opções:

Opção Valor predefinido Description
reliabilityLevel BEST_EFFORT Controla a fiabilidade das operações de inserção. Valores possíveis: BEST_EFFORT (padrão, mais rápido, pode resultar em linhas duplicadas se o executor reiniciar), NO_DUPLICATES (mais lento, garante que não são inseridas linhas duplicadas mesmo que o executor reinicie). Escolha com base na sua tolerância a duplicados e nas necessidades de desempenho.
isolationLevel "READ_COMMITTED" Define o nível de isolamento das transações para operações SQL. Valores possíveis: READ_COMMITTED (por defeito, impede a leitura de dados não comprometidos), READ_UNCOMMITTED, REPEATABLE_READ, SNAPSHOT, SERIALIZABLE. Níveis mais elevados de isolamento podem reduzir a concorrência, mas melhorar a consistência dos dados.
tableLock falso Controla se a dica de bloqueio ao nível de tabela do SQL Server TABLOCK é usada durante as operações de inserção. Valores possíveis: true (ativa o TABLOCK, que pode melhorar o desempenho da escrita em massa), false (por defeito, não usa TABLOCK). Ao definir para true, pode aumentar o débito para grandes inserções, mas pode reduzir a concorrência para outras operações na tabela.
schemaCheckEnabled verdadeiro Controla se a validação rigorosa do esquema é imposta entre o seu Spark DataFrame e a tabela SQL. Valores possíveis: true (por defeito, impõe correspondência estrita de esquemas), false (permite mais flexibilidade e pode saltar algumas verificações de esquema). Definir para false pode ajudar com desajustes de esquema, mas pode levar a resultados inesperados se as estruturas diferirem significativamente.

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

Exemplo de escrita e leitura

O seguinte código utiliza autenticação automática do Microsoft Entra ID para demonstrar estas 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().

Sugestão

Os dados são criados em linha para fins de demonstração. Em um cenário de produção, normalmente leria dados de uma fonte existente ou criaria um DataFrame.

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

Também pode selecionar colunas, aplicar filtros e usar outras opções quando lê dados do motor de base de dados SQL.

Exemplos de autenticação

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

Observação

Como mencionado anteriormente, a autenticação do ID Microsoft Entra é gerida automaticamente quando inicia sessão no espaço de trabalho do Fabric, por isso só precisa de usar estes métodos se o seu cenário assim o exigir.

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

Modos de salvamento DataFrame suportados

Quando escreves dados do Spark para bases de dados SQL, podes escolher entre vários modos de gravação. Os modos de gravação controlam como os dados são escritos quando a tabela de destino já existe, e podem afetar o esquema, os dados e a indexação. Compreender estes modos ajuda-o a evitar perdas ou alterações inesperadas de dados.

Este conector suporta as opções aqui definidas: Funções de Spark Save

  • ErrorIfExists (modo de gravação predefinido): Se a tabela de destino existir, a escrita é abortada e uma exceção é retornada. Caso contrário, é criada uma nova tabela com dados.

  • Ignorar: Se a tabela de destino existir, a escrita ignora o pedido e não devolve o erro. Caso contrário, é criada uma nova tabela com dados.

  • Sobrescrever: Se a tabela de destino existir, a tabela é eliminada, recriada e novos dados são adicionados.

    Observação

    Quando usas overwrite, perdes o esquema original da tabela (especialmente os tipos de dados exclusivos do MSSQL) e os índices das tabelas. O esquema é substituído pelo esquema inferido a partir do seu DataFrame Spark. Para evitar perder o esquema e os índices, adicione .option("truncate", true).

  • Anexar: Se a tabela de destino existir, novos dados são adicionados a ela. Caso contrário, é criada uma nova tabela com dados.

Troubleshoot

Quando o processo é concluído, a saída da sua operação de leitura do Spark aparece na área de saída da célula. Os erros de com.microsoft.sqlserver.jdbc.SQLServerException vêm diretamente do SQL Server. Pode encontrar informações detalhadas sobre erros nos registos de aplicação do Spark.

As escritas em massa exigem que os dados recebidos se ajustem aos tipos de dados SQL alvo. Se os dados não corresponderem, 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, este erro pode ocorrer quando a tabela SQL alvo usa datetime, mas um valor Spark TimestampType 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 que correspondam ao tipo de dados SQL de destino. Por exemplo, trunque o valor para três dígitos com precisão fracionária de segundo: 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

  • Last updated on