Proteja as credenciais com serviços vinculados usando o mssparkutils

O acesso a dados de fontes externas é um padrão comum. A menos que a fonte de dados externa permita acesso anônimo, é provável que você precise proteger sua conexão com uma credencial, um segredo ou uma cadeia de conexão.

O Azure Synapse Analytics usa a passagem do Microsoft Entra por padrão para autenticação entre os recursos. Se você precisar se conectar a um recurso usando outras credenciais, use o mssparkutils diretamente. O pacote mssparkutils simplifica o processo de recuperação dos tokens SAS, tokens do Microsoft Entra, cadeias de conexão e segredos armazenados em um serviço vinculado ou em um Azure Key Vault.

A passagem do Microsoft Entra usa permissões atribuídas a você como usuário no Microsoft Entra ID, em vez de permissões atribuídas ao Synapse ou a uma entidade de serviço separada. Por exemplo, caso queira usar a passagem do Microsoft Entra para acessar um blob em uma conta de armazenamento, deverá ir para essa conta de armazenamento e atribuir a função de colaborador do blob a si mesmo.

Ao recuperar segredos do Azure Key Vault, é recomendável criar um serviço vinculado no seu Azure Key Vault. Verifique se as identidades gerenciadas para recursos do Azure do workspace do Azure Synapse tem privilégios Get do Segredo no seu Azure Key Vault. O Azure Synapse será autenticado no Azure Key Vault usando as identidades gerenciadas para recursos do Azure do workspace do Azure Synapse. Se você se conectar diretamente ao Azure Key Vault sem um serviço vinculado, autentique usando sua credencial de usuário Microsoft Entra.

Para obter mais informações, confira serviços vinculados.

Uso

Ajuda do mssparkutils para tokens e segredos

Essa função exibe a documentação de ajuda para o gerenciamento de segredos e tokens no Synapse.

mssparkutils.credentials.help()

mssparkutils.credentials.help()

Console.WriteLine(TokenLibrary.help());

Obter resultado:

 getToken(audience: String, name: String): returns AAD token for a given audience, name (optional)
 isValidToken(token: String): returns true if token hasn't expired
 getConnectionStringOrCreds(linkedService: String): returns connection string or credentials for the linked service
 getFullConnectionString(linkedService: String): returns full connection string with credentials for the linked service
 getPropertiesAll(linkedService: String): returns all the properties of the linked service
 getSecret(akvName: String, secret: String, linkedService: String): returns AKV secret for a given AKV linked service, akvName, secret key using workspace MSI
 getSecret(akvName: String, secret: String): returns AKV secret for a given akvName, secret key using user credentials
 getSecretWithLS(linkedService: String, secret: String): returns AKV secret for a given linked service, secret key
 putSecret(akvName: String, secretName: String, secretValue: String): puts AKV secret for a given akvName, secretName
 putSecret(akvName: String, secretName: String, secretValue: String, linkedService: String): puts AKV secret for a given akvName, secretName
 putSecretWithLS(linkedService: String, secretName: String, secretValue: String): puts AKV secret for a given linked service, secretName

Acesse o Azure Data Lake Storage Gen2

Armazenamento principal do ADLS Gen2

O acesso a arquivos do Azure Data Lake Storage primário usa a passagem do Microsoft Entra para autenticação por padrão e não requer o uso explícito dos mssparkutils. A identidade usada na autenticação da passagem difere com base em alguns fatores. Por padrão, os notebooks interativos são executados usando a identidade do usuário, mas isso pode ser alterado para a identidade do serviço gerenciado (MSI) do espaço de trabalho. Os trabalhos em lote e as execuções não interativas do bloco de anotações usam a MSI do espaço de trabalho.

val df = spark.read.csv("abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>")
display(df.limit(10))

df = spark.read.csv('abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>')
display(df.limit(10))

Armazenamento do ADLS Gen2 com serviços vinculados

O Azure Synapse Analytics proporciona uma experiência integrada de serviços vinculados ao se conectar ao Azure Data Lake Storage Gen2. Os serviços vinculados podem ser configurados para serem autenticados usando uma Chave de Conta, uma Entidade de Serviço, uma Identidade Gerenciada ou uma Credencial.

Quando o método de autenticação do serviço vinculado é definido como Chave de Conta, o serviço vinculado autentica usando a chave da conta de armazenamento fornecida, solicita uma chave SAS e aplica-a automaticamente à solicitação de armazenamento usando o método LinkedServiceBasedSASProvider.

O Synapse permite que os usuários definam o serviço vinculado para uma conta de armazenamento específica. Isso possibilita a leitura/gravação de dados de várias contas de armazenamento em um único aplicativo/consulta do Spark. Depois de definir spark.storage.synapse.{source_full_storage_account_name}.linkedServiceName para cada conta de armazenamento que será usada, o Synapse descobre qual serviço vinculado usar para uma operação de leitura/gravação específica. No entanto, se nosso trabalho spark lidar apenas com uma única conta de armazenamento, podemos omitir o nome da conta de armazenamento e usar spark.storage.synapse.linkedServiceName.

val sc = spark.sparkContext
val source_full_storage_account_name = "teststorage.dfs.core.windows.net"
spark.conf.set(s"spark.storage.synapse.$source_full_storage_account_name.linkedServiceName", "<LINKED SERVICE NAME>")
sc.hadoopConfiguration.set(s"fs.azure.account.auth.type.$source_full_storage_account_name", "SAS")
sc.hadoopConfiguration.set(s"fs.azure.sas.token.provider.type.$source_full_storage_account_name", "com.microsoft.azure.synapse.tokenlibrary.LinkedServiceBasedSASProvider")

val df = spark.read.csv("abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>")

display(df.limit(10))

%%pyspark
# Set the required configs
source_full_storage_account_name = "teststorage.dfs.core.windows.net"
spark.conf.set(f"spark.storage.synapse.{source_full_storage_account_name}.linkedServiceName", "<lINKED SERVICE NAME>")
sc._jsc.hadoopConfiguration().set(f"fs.azure.account.auth.type.{source_full_storage_account_name}", "SAS")
sc._jsc.hadoopConfiguration().set(f"fs.azure.sas.token.provider.type.{source_full_storage_account_name}", "com.microsoft.azure.synapse.tokenlibrary.LinkedServiceBasedSASProvider")

# Python code
df = spark.read.csv('abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<DIRECTORY PATH>')

df.show()

Quando o método de autenticação de serviço vinculado estiver definido como Identidade Gerenciada ou Entidade de Serviço, o serviço vinculado usará o token de Identidade Gerenciada ou de Entidade de Serviço com o provedor LinkedServiceBasedTokenProvider.

val sc = spark.sparkContext
val source_full_storage_account_name = "teststorage.dfs.core.windows.net"
spark.conf.set(s"spark.storage.synapse.$source_full_storage_account_name.linkedServiceName", "<LINKED SERVICE NAME>")
sc.hadoopConfiguration.set(s"fs.azure.account.oauth.provider.type.$source_full_storage_account_name", "com.microsoft.azure.synapse.tokenlibrary.LinkedServiceBasedTokenProvider") 
val df = spark.read.csv("abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>")

display(df.limit(10))

%%pyspark
# Python code
source_full_storage_account_name = "teststorage.dfs.core.windows.net"
spark.conf.set(f"spark.storage.synapse.{source_full_storage_account_name}.linkedServiceName", "<LINKED SERVICE NAME>")
sc._jsc.hadoopConfiguration().set(f"fs.azure.account.oauth.provider.type.{source_full_storage_account_name}", "com.microsoft.azure.synapse.tokenlibrary.LinkedServiceBasedTokenProvider")

df = spark.read.csv('abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<DIRECTORY PATH>')

df.show()

Definir configurações de autenticação por meio da configuração do Spark

As configurações de autenticação também podem ser especificadas por meio de configurações spark, em vez de executar instruções spark. Todas as configurações do Spark devem ser prefixadas com spark. e todas as configurações do Hadoop devem ser prefixadas com spark.hadoop..

Armazenamento do ADLS Gen2 sem serviços vinculados

Conecte-se diretamente ao armazenamento do ADLS Gen2 usando uma chave de SAS. Use o ConfBasedSASProvider e forneça a chave de SAS para a definição de configuração spark.storage.synapse.sas. ​ Os tokens SAS podem ser definidos no nível do contêiner, no nível da conta ou global. Não é recomendável definir chaves de SAS no nível global, pois o trabalho não poderá fazer a leitura/gravar de mais de uma conta de armazenamento.

Configuração de SAS por contêiner de armazenamento

%%spark
sc.hadoopConfiguration.set("fs.azure.account.auth.type.<ACCOUNT>.dfs.core.windows.net", "SAS")
sc.hadoopConfiguration.set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.ConfBasedSASProvider")
spark.conf.set("spark.storage.synapse.<CONTAINER>.<ACCOUNT>.dfs.core.windows.net.sas", "<SAS KEY>")

val df = spark.read.csv("abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>")

display(df.limit(10))

%%pyspark

sc._jsc.hadoopConfiguration().set("fs.azure.account.auth.type.<ACCOUNT>.dfs.core.windows.net", "SAS")
sc._jsc.hadoopConfiguration().set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.ConfBasedSASProvider")
spark.conf.set("spark.storage.synapse.<CONTAINER>.<ACCOUNT>.dfs.core.windows.net.sas", "<SAS KEY>")

df = spark.read.csv('abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>')

display(df.limit(10))

Configuração de SAS por conta de armazenamento

%%spark
sc.hadoopConfiguration.set("fs.azure.account.auth.type.<ACCOUNT>.dfs.core.windows.net", "SAS")
sc.hadoopConfiguration.set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.ConfBasedSASProvider")
spark.conf.set("spark.storage.synapse.<ACCOUNT>.dfs.core.windows.net.sas", "<SAS KEY>")

val df = spark.read.csv("abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>")

display(df.limit(10))

%%pyspark

sc._jsc.hadoopConfiguration().set("fs.azure.account.auth.type.<ACCOUNT>.dfs.core.windows.net", "SAS")
sc._jsc.hadoopConfiguration().set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.ConfBasedSASProvider")
spark.conf.set("spark.storage.synapse.<ACCOUNT>.dfs.core.windows.net.sas", "<SAS KEY>")

df = spark.read.csv('abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>')

display(df.limit(10))

Configuração de SAS de todas as contas de armazenamento

%%spark
sc.hadoopConfiguration.set("fs.azure.account.auth.type", "SAS")
sc.hadoopConfiguration.set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.ConfBasedSASProvider")
spark.conf.set("spark.storage.synapse.sas", "<SAS KEY>")

val df = spark.read.csv("abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>")

display(df.limit(10))

%%pyspark

sc._jsc.hadoopConfiguration().set("fs.azure.account.auth.type", "SAS")
sc._jsc.hadoopConfiguration().set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.ConfBasedSASProvider")
spark.conf.set("spark.storage.synapse.sas", "<SAS KEY>")

df = spark.read.csv('abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>')

display(df.limit(10))

Use MSAL para adquirir tokens (usando credenciais de aplicativo personalizadas)

Quando o driver de armazenamento ABFS está configurado para usar MSAL diretamente para autenticações, o provedor não armazena tokens em cache. Isso pode resultar em problemas de confiabilidade. Recomendamos usar o ClientCredsTokenProvider que faz parte do Synapse Spark.

%%spark
val source_full_storage_account_name = "teststorage.dfs.core.windows.net"
sc.hadoopConfiguration.set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.id.$source_full_storage_account_name", "<Entra AppId>")
spark.conf.set("fs.azure.account.oauth2.client.secret.$source_full_storage_account_name", "<Entra app secret>")
spark.conf.set("fs.azure.account.oauth2.client.endpoint.$source_full_storage_account_name", "https://login.microsoftonline.com/<tenantid>")

val df = spark.read.csv("abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>")

display(df.limit(10))

%%pyspark
source_full_storage_account_name = "teststorage.dfs.core.windows.net"
sc._jsc.hadoopConfiguration().set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.ClientCredsTokenProvider")
spark.conf.set(f"fs.azure.account.oauth2.client.id.{source_full_storage_account_name}.linkedServiceName", "<Entra AppId>")
spark.conf.set(f"fs.azure.account.oauth2.client.secret.{source_full_storage_account_name}.linkedServiceName", "<Entra app secret>")
spark.conf.set(f"fs.azure.account.oauth2.client.endpoint.{source_full_storage_account_name}.linkedServiceName", "https://login.microsoftonline.com/<tenantid>")

df = spark.read.csv('abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>')
display(df.limit(10))

Armazenamento ADLS Gen2 com token SAS (do Azure Key Vault)

Conecte-se ao armazenamento do ADLS Gen2 usando um token SAS armazenado no segredo do Azure Key Vault.

%%spark
sc.hadoopConfiguration.set("fs.azure.account.auth.type", "SAS")
sc.hadoopConfiguration.set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.AkvBasedSASProvider")
spark.conf.set("spark.storage.synapse.akv", "<AZURE KEY VAULT NAME>")
spark.conf.set("spark.storage.akv.secret", "<SECRET KEY>")

val df = spark.read.csv("abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>")

display(df.limit(10))

%%pyspark
sc._jsc.hadoopConfiguration().set("fs.azure.account.auth.type", "SAS")
sc._jsc.hadoopConfiguration().set("fs.azure.sas.token.provider.type", "com.microsoft.azure.synapse.tokenlibrary.AkvBasedSASProvider")
spark.conf.set("spark.storage.synapse.akv", "<AZURE KEY VAULT NAME>")
spark.conf.set("spark.storage.akv.secret", "<SECRET KEY>")

df = spark.read.csv('abfss://<CONTAINER>@<ACCOUNT>.dfs.core.windows.net/<FILE PATH>')

display(df.limit(10))

TokenLibrary para outros serviços vinculados

Para conectar-se a outros serviços vinculados, você pode fazer uma chamada direta à TokenLibrary.

getConnectionString()

Para recuperar a cadeia de conexão, use a função getConnectionString e passe o nome do serviço vinculado.

%%spark
// retrieve connectionstring from mssparkutils

mssparkutils.credentials.getFullConnectionString("<LINKED SERVICE NAME>")

%%pyspark
# retrieve connectionstring from mssparkutils

mssparkutils.credentials.getFullConnectionString("<LINKED SERVICE NAME>")

%%csharp
// retrieve connectionstring from TokenLibrary

using Microsoft.Spark.Extensions.Azure.Synapse.Analytics.Utils;

string connectionString = TokenLibrary.GetConnectionString(<LINKED SERVICE NAME>);
Console.WriteLine(connectionString);

getPropertiesAll()

A getPropertiesAll é uma função auxiliar disponível no Scala e no Python para obter todas as propriedades de um serviço vinculado

%%pyspark
import json
# retrieve connectionstring from mssparkutils

json.loads(mssparkutils.credentials.getPropertiesAll("<LINKED SERVICE NAME>"))

A saída terá a seguinte aparência

{
    'AuthType': 'Key',
    'AuthKey': '[REDACTED]',
    'Id': None,
    'Type': 'AzureBlobStorage',
    'Endpoint': 'https://storageaccount.blob.core.windows.net/',
    'Database': None
}

GetSecret()

Para recuperar um segredo armazenado no Azure Key Vault, recomendamos que você crie um serviço vinculado ao Azure Key Vault no workspace do Azure Synapse. As identidades gerenciadas para recursos do Azure do workspace do Azure Synapse precisará ter a permissão GET dos Segredos para o Azure Key Vault. O serviço vinculado usará as identidades gerenciadas para recursos do Azure para se conectar ao serviço do Azure Key Vault para recuperar o segredo. Caso contrário, conectar-se diretamente ao Azure Key Vault usará a credencial do Microsoft Entra do usuário. Nesse caso, o usuário precisará receber as permissões Get do Segredo no Azure Key Vault.

Nas nuvens governamentais, forneça o nome de domínio totalmente qualificado do keyvault.

mssparkutils.credentials.getSecret("<AZURE KEY VAULT NAME>", "<SECRET KEY>" [, <LINKED SERVICE NAME>])

Para recuperar um segredo do Azure Key Vault, use a função mssparkutils.credentials.getSecret().

mssparkutils.credentials.getSecret("<AZURE KEY VAULT NAME>", "<SECRET KEY>", "<LINKED SERVICE NAME>")

mssparkutils.credentials.getSecret("<AZURE KEY VAULT NAME>", "<SECRET KEY>", "<LINKED SERVICE NAME>")

using Microsoft.Spark.Extensions.Azure.Synapse.Analytics.Utils;

string connectionString = TokenLibrary.GetSecret("<AZURE KEY VAULT NAME>", "<SECRET KEY>", "<LINKED SERVICE NAME>");
Console.WriteLine(connectionString);

Conexões de serviços vinculados compatíveis com o runtime do Spark

Embora o Azure Synapse Analytics suporte várias conexões de serviço vinculadas (de pipelines e outros produtos do Azure), nem todas elas têm suporte no tempo de execução do Spark. Aqui está a lista de serviços vinculados com suporte:

• Armazenamento do Blobs do Azure
• Serviços de IA do Azure
• Azure Cosmos DB
• Azure Data Explorer
• Banco de Dados do Azure para MySQL
• Banco de Dados do Azure para PostgreSQL
• Azure Data Lake Storage (Gen1)
• Cofre de Chave do Azure
• Azure Machine Learning
• Azure Purview
• Banco de Dados SQL do Azure
• SQL do Azure Data Warehouse (dedicado e sem servidor)
• Armazenamento do Azure

mssparkutils.credentials.getToken()

Quando você precisar de um token de portador OAuth para acessar os serviços diretamente, poderá usar o método getToken. Há suporte para os seguintes recursos:

Acesso de serviço vinculado sem suporte do runtime do Spark

Os seguintes métodos de acesso aos serviços vinculados não são compatíveis com o runtime do Spark:

• Passagem de argumentos para o serviço vinculado parametrizado
• Conexões com identidades gerenciadas atribuídas ao usuário (UAMI)
• Obtendo o token do portador para o recurso Keyvault quando seu Notebook/SparkJobDefinition é executado como identidade gerenciada
  • Como alternativa, em vez de obter um token de acesso, você pode criar um serviço vinculado ao Keyvault e obter o segredo do seu Notebook/trabalho em lote
• Há suporte apenas para acesso baseado em chave, para conexões do Azure Cosmos DB. Não há suporte para acesso baseado em token.

Ao executar um notebook ou um trabalho do Spark, as solicitações para obter um token/segredo usando um serviço vinculado podem falhar com uma mensagem de erro que indica "BadRequest". Isso geralmente é causado por um problema de configuração do serviço vinculado. Se você vir essa mensagem de erro, verifique a configuração do serviço vinculado. Se tiver alguma dúvida, entre em contato com o Suporte do Microsoft Azure no portal do Azure.

Conteúdo relacionado

• Gravar no pool de SQL dedicado
• Apache Spark no Azure Synapse Analytics
• Introdução aos Utilitários do Microsoft Spark