Interagir com o Azure Cosmos DB usando o Apache Spark 2 no Azure Synapse Link
Nota
Para o Azure Synapse Link para Azure Cosmos DB usando o Spark 3, consulte este artigo Azure Synapse Link for Azure Cosmos DB on Spark 3
Neste artigo, você aprenderá a interagir com o Azure Cosmos DB usando o Synapse Apache Spark 2. Com seu suporte total para Scala, Python, SparkSQL e C#, o Synapse Apache Spark é fundamental para cenários de análise, engenharia de dados, ciência de dados e exploração de dados no Azure Synapse Link for Azure Cosmos DB.
Os seguintes recursos são suportados durante a interação com o Azure Cosmos DB:
- O Synapse Apache Spark permite que você analise dados em seus contêineres do Azure Cosmos DB habilitados com o Azure Synapse Link quase em tempo real sem afetar o desempenho de suas cargas de trabalho transacionais. As duas opções a seguir estão disponíveis para consultar o repositório analítico do Azure Cosmos DB do Spark:
- Carregar para o Spark DataFrame
- Criar tabela do Spark
- O Synapse Apache Spark também permite que você ingira dados no Azure Cosmos DB. É importante observar que os dados são sempre ingeridos em contêineres do Azure Cosmos DB por meio do repositório transacional. Quando o Synapse Link está habilitado, todas as novas inserções, atualizações e exclusões são sincronizadas automaticamente com o repositório analítico.
- O Synapse Apache Spark também suporta streaming estruturado do Spark com o Azure Cosmos DB como fonte e coletor.
As seções a seguir orientam você pela sintaxe dos recursos acima. Você também pode conferir o módulo Learn sobre como consultar o Azure Cosmos DB com o Apache Spark for Azure Synapse Analytics. Os gestos no espaço de trabalho do Azure Synapse Analytics são projetados para fornecer uma experiência fácil e pronta para começar. Os gestos ficam visíveis quando você clica com o botão direito do mouse em um contêiner do Azure Cosmos DB na guia Dados do espaço de trabalho Sinapse. Com os gestos, pode gerar rapidamente código e personalizá-lo de acordo com as suas necessidades. Os gestos também são perfeitos para detetar dados com um único clique.
Importante
Você deve estar ciente de algumas restrições no esquema analítico que podem levar ao comportamento inesperado em operações de carregamento de dados. Por exemplo, apenas as primeiras 1000 propriedades do esquema transacional estão disponíveis no esquema analítico, as propriedades com espaços não estão disponíveis, etc. Se você estiver enfrentando alguns resultados inesperados, verifique as restrições do esquema de armazenamento analítico para obter mais detalhes.
Consultar repositório analítico do Azure Cosmos DB
Antes de aprender sobre as duas opções possíveis para consultar o repositório analítico do Azure Cosmos DB, carregando para o Spark DataFrame e criando a tabela do Spark, vale a pena explorar as diferenças de experiência para que você possa escolher a opção que funciona para suas necessidades.
A diferença na experiência está em torno de saber se as alterações de dados subjacentes no contêiner do Azure Cosmos DB devem ser refletidas automaticamente na análise executada no Spark. Quando um DataFrame do Spark é registrado ou uma tabela do Spark é criada no armazenamento analítico de um contêiner, os metadados em torno do instantâneo atual dos dados no repositório analítico são buscados no Spark para um pushdown eficiente da análise subsequente. É importante observar que, como o Spark segue uma política de avaliação preguiçosa, a menos que uma ação seja invocada no Spark DataFrame ou uma consulta SparkSQL seja executada na tabela do Spark, os dados reais não são buscados no repositório analítico do contêiner subjacente.
No caso do carregamento para o DataFrame do Spark, os metadados obtidos são colocados em cache durante a duração da sessão do Spark e, portanto, as ações subsequentes invocadas no DataFrame são avaliadas relativamente ao instantâneo do arquivo analítico no momento da criação do DataFrame.
Por outro lado, no caso da criação de uma tabela do Spark, os metadados do estado do arquivo analítico não são colocados em cache no Spark e são recarregados em cada execução de consulta SQL do Spark na tabela do Spark.
Assim sendo, pode escolher entre o carregamento para o DataFrame do Spark e a criação de uma tabela do Spark consoante queira que a análise do Spark seja avaliada relativamente a um instantâneo fixo do arquivo analítico ou ao instantâneo mais recente do arquivo analítico, respetivamente.
Se suas consultas analíticas tiverem filtros usados com freqüência, você terá a opção de particionar com base nesses campos para um melhor desempenho da consulta. Você pode executar periodicamente o trabalho de particionamento de um bloco de anotações do Azure Synapse Spark para acionar o particionamento no repositório analítico. Esse repositório particionado aponta para a conta de armazenamento principal do ADLS Gen2 vinculada ao seu espaço de trabalho do Azure Synapse. Para saber mais, consulte a introdução ao particionamento personalizado e como configurar artigos de particionamento personalizado.
Nota
Para consultar contas do Azure Cosmos DB para MongoDB, saiba mais sobre a representação do esquema de fidelidade total no repositório analítico e os nomes de propriedade estendida a serem usados.
Nota
Observe que todos os comandos abaixo diferenciam maiúsculas options de minúsculas. Por exemplo, você deve usar Gateway while gateway retornará um erro.
Carregar para o Spark DataFrame
Neste exemplo, você criará um Spark DataFrame que aponta para o repositório analítico do Azure Cosmos DB. Em seguida, você pode executar análises adicionais invocando ações do Spark no DataFrame. Essa operação não afeta o repositório transacional.
A sintaxe em Python seria a seguinte:
# To select a preferred list of regions in a multi-region Azure Cosmos DB account, add .option("spark.cosmos.preferredRegions", "<Region1>,<Region2>")
df = spark.read.format("cosmos.olap")\
.option("spark.synapse.linkedService", "<enter linked service name>")\
.option("spark.cosmos.container", "<enter container name>")\
.load()
A sintaxe equivalente em Scala seria a seguinte:
// To select a preferred list of regions in a multi-region Azure Cosmos DB account, add option("spark.cosmos.preferredRegions", "<Region1>,<Region2>")
val df_olap = spark.read.format("cosmos.olap").
option("spark.synapse.linkedService", "<enter linked service name>").
option("spark.cosmos.container", "<enter container name>").
load()
Criar tabela do Spark
Neste exemplo, você criará uma tabela do Spark que aponta o repositório analítico do Azure Cosmos DB. Em seguida, você pode executar análises adicionais invocando consultas SparkSQL na tabela. Essa operação não afeta o armazenamento transacional nem incorre em qualquer movimentação de dados. Se você decidir excluir essa tabela do Spark, o contêiner subjacente do Azure Cosmos DB e o repositório analítico correspondente não serão afetados.
Esse cenário é conveniente para reutilizar tabelas do Spark por meio de ferramentas de terceiros e fornecer acessibilidade aos dados subjacentes para o tempo de execução.
A sintaxe para criar uma tabela Spark é a seguinte:
%%sql
-- To select a preferred list of regions in a multi-region Azure Cosmos DB account, add spark.cosmos.preferredRegions '<Region1>,<Region2>' in the config options
create table call_center using cosmos.olap options (
spark.synapse.linkedService '<enter linked service name>',
spark.cosmos.container '<enter container name>'
)
Nota
Se você tiver cenários em que o esquema do contêiner subjacente do Azure Cosmos DB muda ao longo do tempo; e se você quiser que o esquema atualizado reflita automaticamente nas consultas na tabela do Spark, você pode conseguir isso definindo a spark.cosmos.autoSchemaMerge opção como true nas opções da tabela do Spark.
Gravar o Spark DataFrame no contêiner do Azure Cosmos DB
Neste exemplo, você gravará um Spark DataFrame em um contêiner do Azure Cosmos DB. Essa operação afetará o desempenho de cargas de trabalho transacionais e consumirá unidades de solicitação provisionadas no contêiner do Azure Cosmos DB ou no banco de dados compartilhado.
A sintaxe em Python seria a seguinte:
# Write a Spark DataFrame into an Azure Cosmos DB container
# To select a preferred list of regions in a multi-region Azure Cosmos DB account, add .option("spark.cosmos.preferredRegions", "<Region1>,<Region2>")
YOURDATAFRAME.write.format("cosmos.oltp")\
.option("spark.synapse.linkedService", "<enter linked service name>")\
.option("spark.cosmos.container", "<enter container name>")\
.option("spark.cosmos.write.upsertEnabled", "true")\
.mode('append')\
.save()
A sintaxe equivalente em Scala seria a seguinte:
// To select a preferred list of regions in a multi-region Azure Cosmos DB account, add option("spark.cosmos.preferredRegions", "<Region1>,<Region2>")
import org.apache.spark.sql.SaveMode
df.write.format("cosmos.oltp").
option("spark.synapse.linkedService", "<enter linked service name>").
option("spark.cosmos.container", "<enter container name>").
option("spark.cosmos.write.upsertEnabled", "true").
mode(SaveMode.Overwrite).
save()
Carregar DataFrame de streaming do contêiner
Neste gesto, você usará o recurso Spark Streaming para carregar dados de um contêiner em um dataframe. Os dados serão armazenados na conta principal do data lake (e no sistema de arquivos) que você conectou ao espaço de trabalho.
Nota
Se você está procurando fazer referência a bibliotecas externas no Synapse Apache Spark, saiba mais aqui. Por exemplo, se você estiver procurando ingerir um Spark DataFrame em um contêiner do Azure Cosmos DB para MongoDB, poderá usar o conector MongoDB para Spark.
Carregar DataFrame de streaming do contêiner do Azure Cosmos DB
Neste exemplo, você usará o recurso de streaming estruturado do Spark para carregar dados de um contêiner do Azure Cosmos DB em um DataFrame de streaming do Spark usando a funcionalidade de feed de alterações no Azure Cosmos DB. Os dados do ponto de verificação usados pelo Spark serão armazenados na conta principal do data lake (e no sistema de arquivos) que você conectou ao espaço de trabalho.
Se a pasta /localReadCheckpointFolder não for criada (no exemplo abaixo), ela será criada automaticamente. Essa operação afetará o desempenho de cargas de trabalho transacionais e consumirá Unidades de Solicitação provisionadas no contêiner do Azure Cosmos DB ou no banco de dados compartilhado.
A sintaxe em Python seria a seguinte:
# To select a preferred list of regions in a multi-region Azure Cosmos DB account, add .option("spark.cosmos.preferredRegions", "<Region1>,<Region2>")
dfStream = spark.readStream\
.format("cosmos.oltp")\
.option("spark.synapse.linkedService", "<enter linked service name>")\
.option("spark.cosmos.container", "<enter container name>")\
.option("spark.cosmos.changeFeed.readEnabled", "true")\
.option("spark.cosmos.changeFeed.startFromTheBeginning", "true")\
.option("spark.cosmos.changeFeed.checkpointLocation", "/localReadCheckpointFolder")\
.option("spark.cosmos.changeFeed.queryName", "streamQuery")\
.load()
A sintaxe equivalente em Scala seria a seguinte:
// To select a preferred list of regions in a multi-region Azure Cosmos DB account, add .option("spark.cosmos.preferredRegions", "<Region1>,<Region2>")
val dfStream = spark.readStream.
format("cosmos.oltp").
option("spark.synapse.linkedService", "<enter linked service name>").
option("spark.cosmos.container", "<enter container name>").
option("spark.cosmos.changeFeed.readEnabled", "true").
option("spark.cosmos.changeFeed.startFromTheBeginning", "true").
option("spark.cosmos.changeFeed.checkpointLocation", "/localReadCheckpointFolder").
option("spark.cosmos.changeFeed.queryName", "streamQuery").
load()
Gravar DataFrame de streaming no contêiner do Azure Cosmos DB
Neste exemplo, você gravará um DataFrame de streaming em um contêiner do Azure Cosmos DB. Essa operação afetará o desempenho de cargas de trabalho transacionais e consumirá Unidades de Solicitação provisionadas no contêiner do Azure Cosmos DB ou no banco de dados compartilhado. Se a pasta /localWriteCheckpointFolder não for criada (no exemplo abaixo), ela será criada automaticamente.
A sintaxe em Python seria a seguinte:
# To select a preferred list of regions in a multi-region Azure Cosmos DB account, add .option("spark.cosmos.preferredRegions", "<Region1>,<Region2>")
# If you are using managed private endpoints for Azure Cosmos DB analytical store and using batch writes/reads and/or streaming writes/reads to transactional store you should set connectionMode to Gateway.
def writeBatchToCosmos(batchDF, batchId):
batchDF.persist()
print("--> BatchId: {}, Document count: {} : {}".format(batchId, batchDF.count(), datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")))
batchDF.write.format("cosmos.oltp")\
.option("spark.synapse.linkedService", "<enter linked service name>")\
.option("spark.cosmos.container", "<enter container name>")\
.option("spark.cosmos.write.upsertEnabled", "true")\
.mode('append')\
.save()
print("<-- BatchId: {}, Document count: {} : {}".format(batchId, batchDF.count(), datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")))
batchDF.unpersist()
streamQuery = dfStream\
.writeStream\
.foreachBatch(writeBatchToCosmos) \
.option("checkpointLocation", "/localWriteCheckpointFolder")\
.start()
streamQuery.awaitTermination()
A sintaxe equivalente em Scala seria a seguinte:
// To select a preferred list of regions in a multi-region Azure Cosmos DB account, add .option("spark.cosmos.preferredRegions", "<Region1>,<Region2>")
// If you are using managed private endpoints for Azure Cosmos DB analytical store and using batch writes/reads and/or streaming writes/reads to transactional store you should set connectionMode to Gateway.
val query = dfStream.
writeStream.
foreachBatch { (batchDF: DataFrame, batchId: Long) =>
batchDF.persist()
batchDF.write.format("cosmos.oltp").
option("spark.synapse.linkedService", "<enter linked service name>").
option("spark.cosmos.container", "<enter container name>").
option("spark.cosmos.write.upsertEnabled", "true").
mode(SaveMode.Overwrite).
save()
println(s"BatchId: $batchId, Document count: ${batchDF.count()}")
batchDF.unpersist()
()
}.
option("checkpointLocation", "/localWriteCheckpointFolder").
start()
query.awaitTermination()