Conector Spark para o Data Warehouse do Microsoft Fabric

O conector do Spark para o Fabric Data Warehouse permite que desenvolvedores do Spark e cientistas de dados acessem e trabalhem com os dados de um data warehouse e do endpoint de análise de SQL de um lakehouse. O conector oferece os seguintes recursos:

  • Você pode trabalhar com dados de um warehouse ou endpoint de análise SQL no mesmo espaço de trabalho ou em vários espaços de trabalho.
  • O endpoint de análise SQL de um lakehouse é detectado automaticamente com base no contexto do espaço de trabalho.
  • O conector tem uma API simplificada do Spark, abstrai a complexidade subjacente e opera com apenas uma linha de código.
  • Ao acessar uma tabela ou uma exibição, o conector aplica os 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 vem pré-instalado no runtime do Fabric, o que elimina a necessidade de instalação separada.

Autenticação

A autenticação do Microsoft Entra é uma abordagem de autenticação integrada. Os usuários fazem login no espaço de trabalho do Fabric, e suas credenciais são automaticamente encaminhadas para o motor SQL para autenticação e autorização. As credenciais são mapeadas automaticamente, e os usuários não precisarão fornecer as opções de configuração específicas.

Observação

O conector do Spark para Fabric Data Warehouse dá suporte apenas à autenticação interativa do usuário do Microsoft Entra. Não há suporte para autenticação de entidade de serviço.

Permissões

Para se conectar ao mecanismo SQL, os usuários precisam de, no mínimo, permissão de Leitura (semelhante à permissão CONNECT no SQL Server) no data warehouse ou no endpoint de análise SQL (no nível do item). Os usuários também precisam de permissões granulares no nível do objeto para ler dados de tabelas ou exibições específicas. Para saber mais, veja Segurança em Fabric Data Warehouse.

Modelos e exemplos de código

Usar uma assinatura de método

O comando a seguir mostra a assinatura do synapsesql método para a solicitação de leitura. O argumento em três partes tableName é necessário para acessar tabelas ou exibições em um data warehouse e no endpoint de análise SQL de um lakehouse. Atualize o argumento com os seguintes nomes, com base no seu cenário:

  • Parte 1: Nome do armazém ou casa do lago.
  • Parte 2: Nome do esquema.
  • Parte 3: Nome da tabela ou visualização.
synapsesql(tableName:String="<Part 1.Part 2.Part 3>") => org.apache.spark.sql.DataFrame

Além de ler diretamente de uma tabela ou visão, esse conector também permite que você especifique uma consulta personalizada ou pass-through, que é passada para o mecanismo SQL, e o resultado é retornado ao Spark.

spark.read.option(Constants.DatabaseName, "<warehouse/lakeshouse name>").synapsesql("<T-SQL Query>") => org.apache.spark.sql.DataFrame

Embora esse conector descubra automaticamente o ponto de extremidade para o warehouse/lakehouse especificado, é possível especificá-lo explicitamente.

//For warehouse
spark.conf.set("spark.datawarehouse.<warehouse name>.sqlendpoint", "<sql endpoint,port>")
//For lakehouse
spark.conf.set("spark.lakehouse.<lakeshouse name>.sqlendpoint", "<sql endpoint,port>")
//Read from table
spark.read.synapsesql("<warehouse/lakeshouse name>.<schema name>.<table or view name>")

Ler dados no mesmo espaço de trabalho

Importante

Execute estas instruções de importação no início do notebook ou antes de começar a usar o conector:

import com.microsoft.spark.fabric
from com.microsoft.spark.fabric.Constants import Constants

O código a seguir é um exemplo para ler dados de uma tabela ou exibição em um DataFrame do Spark:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>")

O código a seguir é um exemplo para ler dados de uma tabela ou exibição em um DataFrame do Spark com um limite de contagem de linhas de 10:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").limit(10)

O código a seguir é um exemplo para ler dados de uma tabela ou exibição em um DataFrame do Spark depois de aplicar um filtro:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").filter("column name == 'value'")

O código a seguir é um exemplo para ler dados de uma tabela ou exibição em um DataFrame do Spark somente para colunas selecionadas:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").select("column A", "Column B")

Ler dados em espaços de trabalho

Para acessar e ler dados de um warehouse ou lakehouse em diferentes workspaces, você pode especificar a ID do workspace onde seu warehouse ou lakehouse está localizado e, em seguida, a ID do item do lakehouse ou warehouse. A linha a seguir mostra um exemplo de leitura de dados de uma tabela ou exibição em um DataFrame do Spark, a partir do warehouse ou lakehouse, com o ID do workspace e o ID do lakehouse/warehouse especificados:

# For lakehouse
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").synapsesql("<lakehouse name>.<schema name>.<table or view name>")
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").option(Constants.LakehouseId, "<lakehouse item id>").synapsesql("<lakehouse name>.<schema name>.<table or view name>")

# For warehouse
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").synapsesql("<warehouse name>.<schema name>.<table or view name>")
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").option(Constants.DatawarehouseId, "<warehouse item id>").synapsesql("<warehouse name>.<schema name>.<table or view name>")

Observação

Ao executar o notebook, por padrão, o conector procura o depósito ou lakehouse especificado no espaço de trabalho do lakehouse anexado ao notebook. Para referenciar um warehouse ou lakehouse em outro workspace, especifique o ID do workspace e o ID do item de lakehouse ou warehouse conforme mostrado acima.

Crie uma tabela do lakehouse com base nos dados de um depósito

Essas linhas de código fornecem um exemplo de como ler dados de uma tabela ou visualização em um DataFrame do Spark e usá-los para criar uma tabela em um lakehouse:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>")
df.write.format("delta").saveAsTable("<Lakehouse table name>")

Gravar dados de um DataFrame do Spark na tabela de data warehouse

Esse conector emprega um processo de gravação em duas fases em uma tabela DW do Fabric. Inicialmente, ele coloca os dados do dataframe do Spark em um armazenamento intermediário, seguido pelo uso do COPY INTO comando para ingerir os dados na tabela DW do Fabric. Essa abordagem garante a escalabilidade com o aumento do volume de dados.

Modos de salvamento de DataFrame com suporte

Há suporte para os seguintes modos de salvamento ao gravar dados de origem de um dataframe em uma tabela de destino no warehouse:

  • ErrorIfExists (modo de salvamento padrão): se a tabela de destino existir, a gravação será anulada com uma exceção retornada ao chamador. Caso contrário, uma nova tabela é criada com dados.
  • Ignorar: se a tabela de destino existir, a gravação ignorará a solicitação de gravação sem retornar um erro. Caso contrário, uma nova tabela é criada com dados.
  • Substituir: se a tabela de destino existir, os dados existentes no destino serão substituídos por dados. Caso contrário, uma nova tabela é criada com dados.
  • Acrescentar: se a tabela de destino existir, os novos dados serão acrescentados a ela. Caso contrário, uma nova tabela é criada com dados.

O código a seguir mostra exemplos de como gravar dados do dataframe do Spark em uma tabela dw do Fabric:

df.write.synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>") # this uses default mode - errorifexists

df.write.mode("errorifexists").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("ignore").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("append").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("overwrite").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")

Observação

O conector suporta gravar somente em uma tabela do Fabric DW, já que o endpoint de análise SQL de um lakehouse é somente leitura.

Paralelizando leituras para melhorar o desempenho

Esse conector dá suporte a leituras paralelizadas para melhorar o desempenho da consulta ao carregar tabelas grandes. Semelhante ao spark.read.jdbc, você pode habilitar o paralelismo especificando uma coluna de partição e seu intervalo de valores. O Spark dividirá a operação de leitura em várias partições que são processadas simultaneamente.

O código a seguir mostra como habilitar leituras paralelas usando um notebook Spark:

import com.microsoft.spark.fabric.tds.implicits.read.FabricSparkTDSImplicits._
import com.microsoft.spark.fabric.tds.implicits.write.FabricSparkTDSImplicits._
import com.microsoft.spark.fabric.Constants
import org.apache.spark.sql.SaveMode 

val df = spark.read.option("partitionColumn", <SomeColumn>)

        .option("lowerBound",  <ColumnValuesLowerLimit>)

        .option("upperBound", <ColumnValuesUpperLimit>)

        .option("numPartitions", <NumberOfPartitionsDesired>).synapsesql(<Table>)

Solucionar problemas

Quando a operação for concluída, o trecho da resposta de leitura será exibido na saída da célula. A falha na célula atual também cancela a execução das células subsequentes do notebook. As informações detalhadas de erros estarão disponíveis nos logs do aplicativo Spark.

Considerações sobre como usar esse conector

Atualmente, o conector:

  • Dá suporte à recuperação ou leitura de dados em armazéns do Fabric e pontos de extremidade analíticos SQL de itens de lakehouse.
  • Suporta gravação de dados em uma tabela de armazenamento usando diferentes modos de salvamento – esse recurso está disponível apenas com o último runtime GA, Runtime 2.0.
  • Com Link Privado habilitado, há suporte para operações de leitura e gravação nos níveis de locatário e workspace. As operações de gravação usam estratégias de inserção em lote com JDBC. Para obter mais informações, consulte suporte do Link Privado para o conector Spark.
  • O Fabric DW agora oferece suporte a Time Travel; no entanto, esse conector não funciona com uma consulta com sintaxe de time travel.
  • Mantém a assinatura de uso como a fornecida com o Apache Spark para Azure Synapse Analytics para consistência. No entanto, não há compatibilidade retroativa para se conectar ao Azure Synapse Analytics e trabalhar com um pool SQL dedicado.
  • Os nomes de coluna com caracteres especiais serão tratados pela adição de caracteres de escape antes do envio da consulta, com base no nome da tabela/visão em três partes. Em uma leitura personalizada ou baseada em consulta de passagem, os usuários são obrigados a escapar de nomes de colunas que contenham caracteres especiais.

Conteúdo relacionado

  • Última atualização em