Interagir com o Azure Cosmos DB usando o Apache Spark 2 no Azure Synapse Link

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.

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.

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

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.

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

Próximos passos

• Exemplos para começar a usar o Azure Synapse Link no GitHub
• Saiba o que é suportado no Azure Synapse Link para Azure Cosmos DB
• Conectar-se ao Synapse Link para Azure Cosmos DB
• Confira o módulo Learn sobre como consultar o Azure Cosmos DB com o Apache Spark for Azure Synapse Analytics.