Conector Spark para Microsoft Fabric Data Warehouse

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

• Você pode trabalhar com dados de um armazenamento de dados ou um endpoint de análise SQL no mesmo espaço de trabalho ou entre vários espaços de trabalho.
• O endpoint de análise SQL de um lakehouse é automaticamente detetado com base no contexto do espaço de trabalho.
• O conector tem uma API Spark simplificada, abstrai a complexidade subjacente e opera com apenas uma linha de código.
• Enquanto se acede a uma tabela ou vista, o conector mantém os 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 vem pré-instalado dentro do 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 utilizadores iniciam sessão no espaço de trabalho do Fabric, e as suas credenciais são automaticamente passadas para o motor SQL para autenticação e autorização. As credenciais são mapeadas automaticamente e os usuários não precisam fornecer opções de configuração específicas.

Permissões

Para se conectarem ao mecanismo SQL, os utilizadores precisam de, pelo menos, permissão de Leitura (semelhante à permissão CONNECT no SQL Server) no armazém ou no ponto de extremidade de análise SQL ao nível de 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, consulte Segurança em Fabric Data Warehouse.

Modelos de código e exemplos

Utilize uma assinatura de método

O comando seguinte mostra a synapsesql assinatura do método para o pedido de leitura. O argumento de três partes tableName é necessário para acessar tabelas ou exibições de um depósito e o ponto de extremidade 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 vista.

synapsesql(tableName:String="<Part 1.Part 2.Part 3>") => org.apache.spark.sql.DataFrame

Além de ler diretamente de uma tabela ou vista, este conector também permite especificar uma consulta personalizada ou de passagem, que é passada para o mecanismo SQL e cujo resultado é devolvido ao Spark.

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

Embora este conector descubra automaticamente o endpoint para o armazém / lakehouse especificado, se quiser especificá-lo explicitamente, poderá fazê-lo.

//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

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

%%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 

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

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 Spark DataFrame 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 apenas para colunas selecionadas:

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

Ler dados entre espaços de trabalho

Para aceder e ler dados de um data warehouse ou lakehouse em diversos espaços de trabalho, pode especificar o ID do espaço de trabalho onde o seu data warehouse ou lakehouse está, e depois o ID do item do lakehouse ou warehouse. A linha a seguir fornece um exemplo de leitura de dados de uma tabela ou exibição em um DataFrame do Spark do armazém ou lakehouse com o ID do espaço de trabalho especificado e o ID do lakehouse/warehouse:

# 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>")

Criar uma tabela lakehouse com base em dados de um armazém

Estas linhas de código fornecem um exemplo para ler dados de uma tabela ou exibição em um DataFrame do Spark e usá-lo para criar uma tabela 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 dataframe do Spark na tabela de depósito

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

Modos de salvamento DataFrame suportados

Os seguintes modos de salvamento são suportados ao gravar dados de origem de um dataframe em uma tabela de destino no depósito:

• ErrorIfExists (modo de salvamento padrão): Se a tabela de destino existir, a gravação será anulada com uma exceção retornada ao destinatário. Caso contrário, uma nova tabela é criada com dados.
• Ignorar: Se a tabela de destino existir, a solicitação de escrita é ignorada 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 anexados a ela. Caso contrário, uma nova tabela é criada com dados.

O código a seguir mostra exemplos de gravação de dados do dataframe do Spark em uma tabela Fabric DW:

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>")

Paralelizando leituras para desempenho melhorado

Este conector suporta leituras paralelizadas para melhorar o desempenho das consultas ao carregar tabelas grandes. Semelhante ao spark.read.jdbc, pode ativar o paralelismo especificando uma coluna de partição e o seu intervalo de valores. O Spark divide então a operação de leitura em várias partições que são processadas em simultâneo.

O código seguinte mostra como ativar leituras paralelas usando um caderno 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>)

Resolver problemas

Após a conclusão, o trecho de resposta de leitura aparece na saída da célula. Uma falha na célula atual do notebook também cancela as execuções subsequentes das células. Informações detalhadas sobre erros estão disponíveis nos logs do aplicativo Spark.

Considerações sobre o uso desse conector

Atualmente, o conector:

• Suporta recuperação de dados ou leitura de armazéns de malha e pontos de extremidade de análise SQL de itens lakehouse.
• Suporta a escrita de dados para uma tabela warehouse usando diferentes modos de gravação – esta funcionalidade está disponível apenas com o mais recente tempo de execução GA, Runtime 2.0.
• Com o Private Link ativado, as operações de leitura e escrita são suportadas tanto ao nível do inquilino como do espaço de trabalho. As operações de escrita utilizam estratégias de inserção por lotes com JDBC. Para mais informações, consulte Private Link suporte para o conector Spark.
• O Fabric DW agora suporta Time Travel , no entanto, esse conector não funciona para uma consulta com sintaxe de viagem no tempo.
• Retém a assinatura de uso como a enviada com o Apache Spark para o Azure Synapse Analytics para garantir consistência. No entanto, não é retrocompatível conectar-se e trabalhar com um pool SQL dedicado no Azure Synapse Analytics.
• Os nomes de colunas com caracteres especiais serão manipulados ao adicionar um carácter de escape antes de a consulta, com base no nome de uma tabela/vista em três partes, ser enviada. Numa leitura personalizada ou baseada em consulta do tipo "passthrough", os utilizadores são obrigados a colocar entre aspas os nomes de colunas que contenham caracteres especiais.

Conteúdos relacionados

• Suporte de ligação privada para o conector Spark
• Ambientes de execução do Apache Spark no Fabric
• Visão geral do monitoramento do Apache Spark
• Segurança em Fabric Data Warehouse
• Permissões granulares do SQL na estrutura