Apache Spark connector: SQL Server e SQL do Azure

O conector Apache Spark para SQL Server e SQL do Azure é um conector de alto desempenho que pode usar para incluir dados transacionais em análise de big data e persistir resultados para consultas ad hoc ou relatórios. Ao usar o conector, pode usar qualquer base de dados SQL, local ou na cloud, como fonte de dados de entrada ou sumidouro de dados de saída para trabalhos Spark.

Esta biblioteca contém o código-fonte do Apache Spark Connector para plataformas SQL Server e SQL do Azure.

O Apache Spark é um mecanismo de análise unificado para processamento de dados em grande escala.

Importa o conector usando a coordenada que corresponde à tua versão do Spark:

Os conectores Spark 3.3.x e Spark 3.4.x são versões beta, e nenhum deles atingiu disponibilidade geral antes do projeto ser arquivado. A versão 1.3.0-BETA não é compatível com o Microsoft JDBC Driver para SQL Server 7.0.1, Spark 2.4 ou Spark 3.0. As notas de lançamento da versão original 1.3.0 indicam que deve importar a versão 1.3.0, mas essa versão nunca foi publicada. Utilize 1.3.0-BETA em substituição.

Também podes construir o conector a partir da fonte ou descarregar o JAR na secção de Lançamentos no GitHub. Para o código-fonte e o histórico de lançamentos, consulte o repositório arquivado do GitHub do conector SQL Spark.

Funcionalidades suportadas

• Suporte para todas as ligações Spark (Scala, Python, R)
• Autenticação básica e suporte ao Key Tab do Active Directory (AD)
• Reordenado o suporte de gravação dataframe
• Suporte para gravação em instância única do SQL Server e pool de dados em clusters de Big Data do SQL Server
• Suporte de conector confiável para instância única do Sql Server

Opções suportadas

O Apache Spark Connector para SQL Server e SQL do Azure suporta as opções definidas no artigo SQL DataSource JDBC .

Além disso, o conector suporta as seguintes opções:

Defina outras opções de cópia em massa no dataframe. O conector passa estas opções para as APIs bulkcopy durante a escrita.

Comparação de desempenho

O Apache Spark Connector para SQL Server e SQL do Azure é até 15x mais rápido do que o conector JDBC genérico para gravar no SQL Server. As características de desempenho variam consoante o tipo, volume de dados, opções usadas e podem mostrar variações entre cada execução. Os resultados de desempenho a seguir são o tempo necessário para sobrescrever uma tabela SQL com 143,9 milhões de linhas em spark dataframe. O Spark dataframe é construído lendo store_sales a tabela HDFS gerada com o benchmark TPCDS do Spark. O tempo de leitura de store_sales a dataframe é excluído. Os resultados são calculados em média ao longo de três corridas.

Configuração

• Configuração do Spark: número_de_executores = 20, memória_do_executor = '1664 m', núcleos_do_executor = 2
• Configuração do Data Gen: scale_factor=50, partitioned_tables=true
• Ficheiro store_sales de dados com número de linhas 143.997.590

Meio Ambiente

• Cluster de Big Data do SQL Server CU5
• master + 6 nós
• Cada nó é um servidor de 5.ª geração, com 512 GB de RAM, NVM de 4 TB por nó e uma placa de rede de 10 Gbps.

Problemas comuns

java.lang.NoClassDefFoundError: com/microsoft/aad/adal4j/AuthenticationException

Este erro ocorre quando usas uma versão mais antiga do mssql driver no teu ambiente Hadoop. O conector agora inclui este driver. Se anteriormente usou o SQL do Azure Connector e instalou manualmente drivers no seu cluster para compatibilidade com a autenticação Microsoft Entra, remova esses drivers.

Para corrigir o erro:

1. Se estiver a usar um ambiente Hadoop genérico, verifique e remova o mssql JAR com o seguinte comando: rm $HADOOP_HOME/share/hadoop/yarn/lib/mssql-jdbc-6.2.1.jre7.jar. Se estiveres a usar Databricks, adiciona um script global ou cluster init para remover versões antigas do mssql driver da /databricks/jars pasta, ou adiciona esta linha a um script existente: rm /databricks/jars/*mssql*

2. Adicione os pacotes adal4j e mssql. Por exemplo, você pode usar o Maven, mas qualquer maneira deve funcionar.

   Atenção

   Não instale o conector SQL do Spark desta forma.

3. Adicione a classe de driver à sua configuração de conexão. Por exemplo:

   connectionProperties = {
     `Driver`: `com.microsoft.sqlserver.jdbc.SQLServerDriver`
   }`
   

Para mais informações, consulte a resolução em https://github.com/microsoft/sql-spark-connector/issues/26.

Introdução

O Apache Spark Connector para SQL Server e SQL do Azure baseia-se na Spark DataSourceV1 API e SQL Server Bulk API. Utiliza a mesma interface do conector Spark-SQL JDBC incorporado. Ao usar esta integração, pode facilmente integrar o conector e migrar os seus trabalhos Spark existentes atualizando o parâmetro de formato com com.microsoft.sqlserver.jdbc.spark.

Para incluir o conector nos seus projetos, descarregue este repositório e construa o JAR usando SBT.

Escrever numa nova tabela SQL

Se usares o modo overwrite sem essa opção truncate ao recriar a tabela, a operação remove índices. Além disso, uma tabela columnstore é convertida em uma tabela heap. Para manter índices existentes, defina a truncate opção para true. Por exemplo, .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)

Anexar à tabela 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)

Especificar o nível de isolamento

Este conector utiliza o READ_COMMITTED nível de isolamento por padrão quando insere dados em massa na base de dados. Para anular o nível de isolamento, use a mssqlIsolationLevel opção:

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

Leia a partir de uma tabela SQL

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

Autenticação do Microsoft Entra

Exemplo de Python com principal de serviço

context = adal.AuthenticationContext(authority)
token = context.acquire_token_with_client_credentials(resource_app_id_url, service_principal_id, service_principal_secret)
access_token = token["accessToken"]

jdbc_db = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("accessToken", access_token) \
        .option("encrypt", "true") \
        .option("hostNameInCertificate", "*.database.windows.net") \
        .load()

Exemplo de Python com palavra-passe do Active Directory

jdbc_df = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("authentication", "ActiveDirectoryPassword") \
        .option("user", user_name) \
        .option("password", password) \
        .option("encrypt", "true") \
        .option("hostNameInCertificate", "*.database.windows.net") \
        .load()

Para autenticar usando o Active Directory, instale a dependência necessária.

Quando usa ActiveDirectoryPassword, o user valor deve estar no formato UPN, como username@domainname.com.

Para o Scala, instala o com.microsoft.aad.adal4j artefacto.

Para Python, instala a adal biblioteca. Esta biblioteca está disponível através do pip.

Para exemplos, veja os cadernos de exemplo.

Conteúdo relacionado

• Repositório GitHub do conector SQL Spark