Чтение данных из Azure Cosmos DB для таблиц Apache Cassandra с помощью Spark

Область применения: Кассандра

В этой статье описывается, как считывать данные, хранящиеся в Azure Cosmos DB для Apache Cassandra из Spark.

API для конфигурации Cassandra

Задайте следующую конфигурацию Spark в кластере записных книжек. Это разовое действие.

//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

Чтение таблицы с помощью команды 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

Чтение таблицы, с помощью spark.read.cassandraFormat

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

Чтение определенных столбцов в таблице

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

Применить фильтры

Вы можете установить предикаты в базу данных, чтобы обеспечить оптимизацию запросов Spark. Предикат — это условие в запросе, возвращающее значение true или false, которое обычно находится в предложении WHERE. Предикат фильтрует данные в запросе базы данных, уменьшая количество записей, полученных из базы данных и повышая производительность запросов. По умолчанию API набора данных Spark автоматически устанавливает допустимые предложения WHERE в базу данных.

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

Cassandra FiltersРаздел физического плана включает установленный фильтр.

API RRD

Чтение таблицы

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

Чтение определенных столбцов в таблице

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

Представления SQL

Создание временного представления из кадра данных

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

Выполнение запросов к представлению

select * from books_vw where book_pub_year > 1891

Следующие шаги

Ниже приведены дополнительные статьи о работе с Azure Cosmos DB для Apache Cassandra из Spark:

• Операции upsert
• Операции удаления
• Операции агрегирования
• Операции копирования таблиц