Ler dados do Azure Cosmos DB para tabelas Apache Cassandra usando o Spark

APLICA-SE A: Cassandra

Este artigo descreve como ler dados armazenados no Azure Cosmos DB para Apache Cassandra do Spark.

Configuração da API para Cassandra

Defina abaixo a configuração de faísca no cluster de blocos de anotações. É uma atividade única.

//Connection-related
 spark.cassandra.connection.host  YOUR_ACCOUNT_NAME.cassandra.cosmosdb.azure.com  
 spark.cassandra.connection.port  10350  
 spark.cassandra.connection.ssl.enabled  true  
 spark.cassandra.auth.username  YOUR_ACCOUNT_NAME  
 spark.cassandra.auth.password  YOUR_ACCOUNT_KEY  
// if using Spark 2.x
// spark.cassandra.connection.factory  com.microsoft.azure.cosmosdb.cassandra.CosmosDbConnectionFactory  

//Throughput-related...adjust as needed
 spark.cassandra.output.batch.size.rows  1  
// spark.cassandra.connection.connections_per_executor_max  10   // Spark 2.x
 spark.cassandra.connection.remoteConnectionsPerExecutor  10   // Spark 3.x
 spark.cassandra.output.concurrent.writes  1000  
 spark.cassandra.concurrent.reads  512  
 spark.cassandra.output.batch.grouping.buffer.size  1000  
 spark.cassandra.connection.keep_alive_ms  600000000  

API DataFrame

Ler tabela usando o comando session.read.format

import org.apache.spark.sql.cassandra._
//Spark connector
import com.datastax.spark.connector._
import com.datastax.spark.connector.cql.CassandraConnector

//if using Spark 2.x, CosmosDB library for multiple retry
//import com.microsoft.azure.cosmosdb.cassandra

val readBooksDF = sqlContext
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map( "table" -> "books", "keyspace" -> "books_ks"))
  .load

readBooksDF.explain
readBooksDF.show

Ler tabela usando spark.read.cassandraFormat

val readBooksDF = spark.read.cassandraFormat("books", "books_ks", "").load()

Ler colunas específicas na tabela

val readBooksDF = spark
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map( "table" -> "books", "keyspace" -> "books_ks"))
  .load
  .select("book_name","book_author", "book_pub_year")

readBooksDF.printSchema
readBooksDF.explain
readBooksDF.show

Aplicar filtros

Você pode empurrar predicados para o banco de dados para permitir consultas Spark melhor otimizadas. Um predicado é uma condição em uma consulta que retorna true ou false, normalmente localizada na cláusula WHERE. Um push down de predicado filtra os dados na consulta do banco de dados, reduzindo o número de entradas recuperadas do banco de dados e melhorando o desempenho da consulta. Por padrão, a API do Spark Dataset enviará automaticamente cláusulas WHERE válidas para o banco de dados.

val df = spark.read.cassandraFormat("books", "books_ks").load
df.explain
val dfWithPushdown = df.filter(df("book_pub_year") > 1891)
dfWithPushdown.explain

readBooksDF.printSchema
readBooksDF.explain
readBooksDF.show

A Cassandra Filters seção do plano físico inclui o filtro empurrado para baixo.

RDD API

Ler tabela

val bookRDD = sc.cassandraTable("books_ks", "books")
bookRDD.take(5).foreach(println)

Ler colunas específicas na tabela

val booksRDD = sc.cassandraTable("books_ks", "books").select("book_id","book_name").cache
booksRDD.take(5).foreach(println)

Visualizações SQL

Criar uma vista temporária a partir de um quadro de dados

spark
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map( "table" -> "books", "keyspace" -> "books_ks"))
  .load.createOrReplaceTempView("books_vw")

Executar consultas no modo de exibição

select * from books_vw where book_pub_year > 1891

Próximos passos

A seguir estão artigos adicionais sobre como trabalhar com o Azure Cosmos DB para Apache Cassandra do Spark:

• Operações Upsert
• Excluir operações
• Operações de agregação
• Operações de cópia de tabela