Tutorial: Conectar-se ao Azure Cosmos DB for NoSQL usando o Spark

Artigo
06/05/2024

APLICA-SE A: NoSQL

Neste tutorial, você usará o conector do Azure Cosmos DB Spark para ler ou gravar dados de uma conta do Azure Cosmos DB for NoSQL. Este tutorial usa o Azure Databricks e um Jupyter notebook para ilustrar como se integrar à API para NoSQL por meio do Spark. Este tutorial se concentra no Python e no Scala, embora você possa usar qualquer linguagem ou interface com suporte do Spark.

Neste tutorial, você aprenderá a:

Conecte-se a uma API da conta NoSQL usando o Spark e um notebook Jupyter.
Criar recursos de banco de dados e contêiner.
Ingira dados no contêiner.
Consulte dados no contêiner.
Execute operações comuns em itens no contêiner.

Pré-requisitos

Uma conta existente do Azure Cosmos DB for NoSQL.
- Se você tiver uma assinatura existente do Azure, crie uma nova conta.
- Nenhuma assinatura do Azure? Você pode experimentar o Azure Cosmos DB gratuitamente sem necessidade de cartão de crédito.
Um workspace existente do Azure Databricks.

Conectar-se usando o Spark e o Jupyter

Use seu workspace existente do Azure Databricks para criar um cluster de cálculo pronto para usar o Apache Spark 3.4.x, a fim de se conectar à sua conta do Azure Cosmos DB for NoSQL.

Abra o workspace do Azure Databricks.
Na interface do workspace, crie um cluster. Defina o cluster com estas configurações, no mínimo:

Versão Valor

Versão de runtime 13.3 LTS (Scala 2.12, Spark 3.4.1)
Use a interface do workspace para pesquisar pacotes do Maven na Central do Maven com uma ID de grupo de com.azure.cosmos.spark. Instale o pacote especificamente para o Spark 3.4 com uma ID de artefato prefixado com azure-cosmos-spark_3-4 no cluster.
Por fim, crie um notebook.

Dica

Por padrão, o notebook é anexado ao cluster criado recentemente.

Versão	Valor
Versão de runtime	13.3 LTS (Scala 2.12, Spark 3.4.1)

No notebook, defina as configurações do OLTP (processamento de transações online) para o ponto de extremidade da conta NoSQL, o nome do banco de dados e o nome do contêiner.

# Set configuration settings
config = {
  "spark.cosmos.accountEndpoint": "<nosql-account-endpoint>",
  "spark.cosmos.accountKey": "<nosql-account-key>",
  "spark.cosmos.database": "cosmicworks",
  "spark.cosmos.container": "products"
}

# Set configuration settings
val config = Map(
  "spark.cosmos.accountEndpoint" -> "<nosql-account-endpoint>",
  "spark.cosmos.accountKey" -> "<nosql-account-key>",
  "spark.cosmos.database" -> "cosmicworks",
  "spark.cosmos.container" -> "products"
)

Criar um banco de dados e um contêiner

Use a API do Catálogo para gerenciar recursos de conta, como bancos de dados e contêineres. Em seguida, use o OLTP para gerenciar os dados nos recursos do contêiner.

Configure a API do Catálogo para gerenciar a API para recursos NoSQL usando o Spark.

# Configure Catalog Api    
spark.conf.set("spark.sql.catalog.cosmosCatalog", "com.azure.cosmos.spark.CosmosCatalog")
spark.conf.set("spark.sql.catalog.cosmosCatalog.spark.cosmos.accountEndpoint", config["spark.cosmos.accountEndpoint"])
spark.conf.set("spark.sql.catalog.cosmosCatalog.spark.cosmos.accountKey", config["spark.cosmos.accountKey"])

// Configure Catalog Api  
spark.conf.set(s"spark.sql.catalog.cosmosCatalog", "com.azure.cosmos.spark.CosmosCatalog")
spark.conf.set(s"spark.sql.catalog.cosmosCatalog.spark.cosmos.accountEndpoint", config("spark.cosmos.accountEndpoint"))
spark.conf.set(s"spark.sql.catalog.cosmosCatalog.spark.cosmos.accountKey", config("spark.cosmos.accountKey"))

Crie um novo banco de dados chamado cosmicworks usando CREATE DATABASE IF NOT EXISTS.

# Create a database by using the Catalog API    
spark.sql(f"CREATE DATABASE IF NOT EXISTS cosmosCatalog.cosmicworks;")

// Create a database by using the Catalog API  
spark.sql(s"CREATE DATABASE IF NOT EXISTS cosmosCatalog.cosmicworks;")

Crie um novo contêiner chamado products usando CREATE TABLE IF NOT EXISTS. Certifique-se de definir o caminho da chave de partição para /category e habilitar a taxa de transferência de escala automática com uma taxa de transferência máxima de 1000 RUs (unidades de solicitação) por segundo.

# Create a products container by using the Catalog API
spark.sql(("CREATE TABLE IF NOT EXISTS cosmosCatalog.cosmicworks.products USING cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/category', autoScaleMaxThroughput = '1000')"))

// Create a products container by using the Catalog API
spark.sql(("CREATE TABLE IF NOT EXISTS cosmosCatalog.cosmicworks.products USING cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/category', autoScaleMaxThroughput = '1000')"))

Crie outro contêiner chamado employees usando uma configuração de chave de partição hierárquica. Use /organization, /department e /team como o conjunto de caminhos de chave de partição. Siga essa ordem específica. Além disso, defina a taxa de transferência como uma quantidade manual de 400 RUs.

# Create an employees container by using the Catalog API
spark.sql(("CREATE TABLE IF NOT EXISTS cosmosCatalog.cosmicworks.employees USING cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/organization,/department,/team', manualThroughput = '400')"))

// Create an employees container by using the Catalog API
spark.sql(("CREATE TABLE IF NOT EXISTS cosmosCatalog.cosmicworks.employees USING cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/organization,/department,/team', manualThroughput = '400')"))

Execute as células do notebook para validar se o banco de dados e os contêineres foram criados em sua API da conta NoSQL.

Ingerir dados

Crie um conjunto de dados de exemplo. Em seguida, use o OLTP para ingerir esses dados na API do contêiner NoSQL.

Crie um conjunto de dados de exemplo.

# Create sample data    
products = (
  ("68719518391", "gear-surf-surfboards", "Yamba Surfboard", 12, 850.00, False),
  ("68719518371", "gear-surf-surfboards", "Kiama Classic Surfboard", 25, 790.00, True)
)

// Create sample data
val products = Seq(
  ("68719518391", "gear-surf-surfboards", "Yamba Surfboard", 12, 850.00, false),
  ("68719518371", "gear-surf-surfboards", "Kiama Classic Surfboard", 25, 790.00, true)
)

Use spark.createDataFrame e a configuração do OLTP salva anteriormente para adicionar dados de exemplo ao contêiner de destino.

# Ingest sample data    
spark.createDataFrame(products) \
  .toDF("id", "category", "name", "quantity", "price", "clearance") \
  .write \
  .format("cosmos.oltp") \
  .options(**config) \
  .mode("APPEND") \
  .save()

// Ingest sample data
spark.createDataFrame(products)
  .toDF("id", "category", "name", "quantity", "price", "clearance")
  .write
  .format("cosmos.oltp")
  .options(config)
  .mode("APPEND")
  .save()

Consultar dados

Carregue os dados OLTP em um dataframe para executar consultas comuns nos dados. Você pode usar várias sintaxes para filtrar ou consultar dados.

Use spark.read para carregar os dados OLTP em um objeto de dataframe. Use a mesma configuração usada anteriormente neste tutorial. Além disso, defina spark.cosmos.read.inferSchema.enabled como true para permitir que o conector Spark deduza o esquema por meio da amostragem de itens existentes.

# Load data    
df = spark.read.format("cosmos.oltp") \
  .options(**config) \
  .option("spark.cosmos.read.inferSchema.enabled", "true") \
  .load()

// Load data
val df = spark.read.format("cosmos.oltp")
  .options(config)
  .option("spark.cosmos.read.inferSchema.enabled", "true")
  .load()

Renderize o esquema dos dados carregados no dataframe usando printSchema.

# Render schema    
df.printSchema()

// Render schema    
df.printSchema()

Renderize as linhas de dados em que a coluna quantity é menor que 20. Use as funções where e show para executar essa consulta.

# Render filtered data    
df.where("quantity < 20") \
  .show()

// Render filtered data
df.where("quantity < 20")
  .show()

Renderize a primeira linha de dados em que a coluna clearance é true. Use a função filter para executar essa consulta.

# Render 1 row of flitered data    
df.filter(df.clearance == True) \
  .show(1)

// Render 1 row of flitered data
df.filter($"clearance" === true)
  .show(1)

Renderize cinco linhas de dados sem filtro ou truncamento. Use a função show para personalizar a aparência e o número de linhas renderizadas.

# Render five rows of unfiltered and untruncated data    
df.show(5, False)

// Render five rows of unfiltered and untruncated data    
df.show(5, false)

Consulte seus dados usando esta cadeia de consulta NoSQL bruta: SELECT * FROM cosmosCatalog.cosmicworks.products WHERE price > 800

# Render results of raw query    
rawQuery = "SELECT * FROM cosmosCatalog.cosmicworks.products WHERE price > 800"
rawDf = spark.sql(rawQuery)
rawDf.show()

// Render results of raw query    
val rawQuery = s"SELECT * FROM cosmosCatalog.cosmicworks.products WHERE price > 800"
val rawDf = spark.sql(rawQuery)
rawDf.show()

Executar operações comuns

Ao trabalhar com a API para dados NoSQL no Spark, você pode executar atualizações parciais ou trabalhar com dados como JSON bruto.

Para executar uma atualização parcial de um item:

Copie a variável de configuração config existente e modifique as propriedades na nova cópia. Especificamente, configure a estratégia de gravação para ItemPatch. Em seguida, desabilite o suporte em massa. Defina as colunas e as operações mapeadas. Por fim, defina o tipo de operação padrão como Set.

# Copy and modify configuration
configPatch = dict(config)
configPatch["spark.cosmos.write.strategy"] = "ItemPatch"
configPatch["spark.cosmos.write.bulk.enabled"] = "false"
configPatch["spark.cosmos.write.patch.defaultOperationType"] = "Set"
configPatch["spark.cosmos.write.patch.columnConfigs"] = "[col(name).op(set)]"

// Copy and modify configuration
val configPatch = scala.collection.mutable.Map.empty ++ config
configPatch ++= Map(
  "spark.cosmos.write.strategy" -> "ItemPatch",
  "spark.cosmos.write.bulk.enabled" -> "false",
  "spark.cosmos.write.patch.defaultOperationType" -> "Set",
  "spark.cosmos.write.patch.columnConfigs" -> "[col(name).op(set)]"
)

Crie variáveis para a chave de partição de item e o identificador exclusivo que você pretende ter como destino como parte desta operação de patch.

# Specify target item id and partition key
targetItemId = "68719518391"
targetItemPartitionKey = "gear-surf-surfboards"

// Specify target item id and partition key
val targetItemId = "68719518391"
val targetItemPartitionKey = "gear-surf-surfboards"

Crie um conjunto de objetos de patch para especificar o item de destino e os campos que devem ser modificados.

# Create set of patch diffs
patchProducts = [{ "id": f"{targetItemId}", "category": f"{targetItemPartitionKey}", "name": "Yamba New Surfboard" }]

// Create set of patch diffs
val patchProducts = Seq(
  (targetItemId, targetItemPartitionKey, "Yamba New Surfboard")
)

Crie um dataframe usando o conjunto de objetos de patch. Use write para executar a operação de patch.

# Create data frame
spark.createDataFrame(patchProducts) \
  .write \
  .format("cosmos.oltp") \
  .options(**configPatch) \
  .mode("APPEND") \
  .save()

// Create data frame
patchProducts
  .toDF("id", "category", "name")
  .write
  .format("cosmos.oltp")
  .options(configPatch)
  .mode("APPEND")
  .save()

Execute uma consulta para analisar os resultados da operação de patch. O item agora será nomeado Yamba New Surfboard sem outras alterações.

# Create and run query
patchQuery = f"SELECT * FROM cosmosCatalog.cosmicworks.products WHERE id = '{targetItemId}' AND category = '{targetItemPartitionKey}'"
patchDf = spark.sql(patchQuery)
patchDf.show(1)

// Create and run query
val patchQuery = s"SELECT * FROM cosmosCatalog.cosmicworks.products WHERE id = '$targetItemId' AND category = '$targetItemPartitionKey'"
val patchDf = spark.sql(patchQuery)
patchDf.show(1)

Para trabalhar com dados JSON brutos:

Copie a variável de configuração config existente e modifique as propriedades na nova cópia. Especificamente, altere o contêiner de destino para employees. Em seguida, configure a coluna/campo contacts para usar dados JSON brutos.

# Copy and modify configuration
configRawJson = dict(config)
configRawJson["spark.cosmos.container"] = "employees"
configRawJson["spark.cosmos.write.patch.columnConfigs"] = "[col(contacts).path(/contacts).op(set).rawJson]"

// Copy and modify configuration
val configRawJson = scala.collection.mutable.Map.empty ++ config
configRawJson ++= Map(
  "spark.cosmos.container" -> "employees",
  "spark.cosmos.write.patch.columnConfigs" -> "[col(contacts).path(/contacts).op(set).rawJson]"
)

Crie um conjunto de funcionários para ingerir no contêiner.

# Create employee data
employees = (
  ("63476388581", "CosmicWorks", "Marketing", "Outside Sales", "Alain Henry",  '[ { "type": "phone", "value": "425-555-0117" }, { "email": "alain@adventure-works.com" } ]'), 
)

// Create employee data
val employees = Seq(
  ("63476388581", "CosmicWorks", "Marketing", "Outside Sales", "Alain Henry",  """[ { "type": "phone", "value": "425-555-0117" }, { "email": "alain@adventure-works.com" } ]""")
)

Crie um dataframe e use write para ingerir os dados do funcionário.

# Ingest data
spark.createDataFrame(employees) \
  .toDF("id", "organization", "department", "team", "name", "contacts") \
  .write \
  .format("cosmos.oltp") \
  .options(**configRawJson) \
  .mode("APPEND") \
  .save()

// Ingest data
spark.createDataFrame(employees)
  .toDF("id", "organization", "department", "team", "name", "contacts")
  .write
  .format("cosmos.oltp")
  .options(configRawJson)
  .mode("APPEND")
  .save()

Renderize os dados do dataframe usando show. Observe que a coluna contacts é um JSON bruto na saída.

# Read and render data
rawJsonDf = spark.read.format("cosmos.oltp") \
  .options(**configRawJson) \
  .load()
rawJsonDf.show()

// Read and render data
val rawJsonDf = spark.read.format("cosmos.oltp")
  .options(configRawJson)
  .load()
rawJsonDf.show()

Próxima etapa

Conector do Azure Cosmos DB Spark no Repositório Central Maven

Compartilhar via

Tutorial: Conectar-se ao Azure Cosmos DB for NoSQL usando o Spark

Pré-requisitos

Conectar-se usando o Spark e o Jupyter

Criar um banco de dados e um contêiner

Ingerir dados

Consultar dados

Executar operações comuns

Próxima etapa

Recursos adicionais

Compartilhar via

Tutorial: Conectar-se ao Azure Cosmos DB for NoSQL usando o Spark

Pré-requisitos

Conectar-se usando o Spark e o Jupyter

Criar um banco de dados e um contêiner

Ingerir dados

Consultar dados

Executar operações comuns

Conteúdo relacionado

Próxima etapa

Recursos adicionais