Sparkコンテキストの使用

cassandraTablesを読み込むためのPythonサポート

Pythonでは、Sparkストリーミング・コンテキストからのcassandraTablesの読み込みと、データベースへのDStreamの保存をサポートしています。

カラム値の読み取り

CassandraRowオブジェクトのgetメソッドを使用して、テーブルのカラムを読み取ることができます。getメソッドは、カラム名またはカラム・インデックスに基づいて、個々のカラム値にアクセスします。型変換はその場で適用されます。null値が返されることが予想される場合は、getOptionバリアントを使用します。

コレクションの読み取り

CassandraRowオブジェクトのgetメソッドを使用して、テーブルのコレクション・カラムを読み取ることができます。getメソッドは、コレクション・カラムにアクセスし、対応するScalaコレクションを返します。

testキースペースが既にセットアップされているという前提で、以下の手順に従ってコレクションにアクセスします。

1. testキースペースで、cqlshを使用してコレクション・セットをセットアップします。

   CREATE TABLE test.users (
     username text PRIMARY KEY, emails SETtext);
   
   INSERT INTO test.users (username, emails) 
     VALUES ('someone', {'someone@email.com', 's@email.com'});

2. Sparkが実行されていない場合は、Sparkシェルを起動します。シェルを起動するためにsudoを使用しないでください。

   bin/dse spark

   「Welcome to Spark」という出力とプロンプトが表示されます。

3. コレクション・セットにアクセスするためのCassandraRDD[CassandraRow]を定義します。

   val row = sc.cassandraTable("test", "users").toArray.apply(0)
   

   row: com.datastax.spark.connector.CassandraRow = CassandraRow{username: someone, emails: {s@email.com,someone@email.com}}

4. コレクション・セットをSparkからクエリーします。

   row.getList[String]("emails")
   

   res2: Vector[String] = Vector(s@email.com, someone@email.com)

   row.get[List[String]]("emails")
   

   res3: List[String] = List(s@email.com, someone@email.com)

   row.get[Seq[String]]("emails")
   

   res4: Seq[String] = List(s@email.com, someone@email.com)

   row.get[IndexedSeq[String]]("emails")
   

   res5: IndexedSeq[String] = Vector(s@email.com, someone@email.com)

   row.get[Set[String]]("emails")
   

   res6: Set[String] = Set(s@email.com, someone@email.com)

   row.get[String]("emails")
   

   res7: String = {s@email.com,someone@email.com}

フェッチ対象カラムの数の制限

パフォーマンス上の理由のため、不要なカラムはフェッチ対象から除外する必要があります。これを行うには、selectメソッドを使用します。

フェッチ対象カラムの数を制限するには：

val row = sc.cassandraTable("test", "users").select("username").toArray

row: Array[com.datastax.spark.connector.CassandraRow] = Array(CassandraRow{username: someone})

行とタプルおよびcaseクラス間のマッピング

行をCassandraRowクラスのオブジェクトにマッピングする代わりに、必要な型のタプルにカラム値を直接ラップ解除することができます。

行をタプルにマッピングするには：

sc.cassandraTable[(String, Int)]("test", "words").select("word", "count").toArray

res9: Array[(String, Int)] = Array((bar,20), (foo,10))

sc.cassandraTable[(Int, String)]("test", "words").select("count", "word").toArray

res10: Array[(Int, String)] = Array((20,bar), (10,foo))

カラムと同じ名前のプロパティを使用してcaseクラスを定義します。複数の単語で構成されるカラムIDの場合は、カラムを作成する際、各単語をアンダースコアで区切り、Scala側でキャメル・ケース略語を使用します。

行をcaseクラスにマッピングするには：

case class WordCount(word: String, count: Int)

defined class WordCount

scala> sc.cassandraTable[WordCount]("test", "words").toArray

res14: Array[WordCount] = Array(WordCount(bar,20), WordCount(foo,20))

カラム名は、以下の規則に従って付けることができます。

• アンダースコアと小文字を使用します。（推奨）
• キャメル・ケースを使用し、Scalaのプロパティと完全に一致させます。

以下の例に有効なカラム名を示します。

ユーザー定義の関数を使用した行とオブジェクト間のマッピング

asをCassandraRDDに対して呼び出して、すべての行を異なる型のオブジェクトにマッピングします。mapとは対照的に、asでは、関数の引数の数がフェッチ対象カラムの数と同じである必要があります。この方法でasを呼び出すと、型変換が実行されます。asを使用して特定の型のオブジェクトを直接作成すると、CassandraRowオブジェクトを作成する必要がなくなり、また、ガーベージ・コレクションの負荷も軽減されます。

ユーザー定義の関数を使用してカラムをマッピングするには：

val table = sc.cassandraTable("test", "words")

table: com.datastax.spark.connector.rdd.CassandraRDD[com.datastax.spark.connector.CassandraRow] = CassandraRDD[9] at RDD at CassandraRDD.scala:47

val total = table.select("count").as((c: Int) => c).sum

total: Double = 30.0

val frequencies = table.select("word", "count").as((w: String, c: Int) => (w, c / total)).toArray

frequencies: Array[(String, Double)] = Array((bar,0.6666666666666666), (foo,0.3333333333333333))

サーバー上の行のフィルター

行をフィルターするには、Sparkのフィルター変換機能を使用します。フィルター変換機能は、まずデータベースからすべての行をフェッチしてから、それをSparkでフィルターします。一部のCPUサイクルは、結果から除外されたオブジェクトのシリアライズとデシリアライズのために浪費されます。このオーバーヘッドを防ぐため、CassandraRDDには、サーバー上の行セットをフィルターするための任意のCQL条件を渡すメソッドが用意されています。