スパークCanssandraデータのためのcsvファイルを生成するためのスクリプト

Question

私は 'csv'ファイルをcassandraのテーブルの以下のロジックに従って生成します。

val df = sc.parallelize(Seq(("a",1,"abc@gmail.com"), ("b",2,"def@gmail.com"),("a",1,"xyz@gmail.com"),("a",2,"abc@gmail.com"))).toDF("col1","col2","emailId")

私はロジック下記のとおり 'CSV' ファイルを生成したいです。 3つの異なる 'emailid'があるので、3つの異なる 'csv'ファイルを生成する必要があります。

3つの異なるクエリのための3つのcsvファイル。

select * from table where emailId='abc@gmail.com' select * from table where emailId='def@gmail.com' select * from table where emailId='xyz@gmail.com' どうすればいいですか？誰でもこのことを助けてくれますか？

バージョン： スパーク1.6.2 Scalaの2.10

Answer 1

は、それらを反復処理する電子メールの明確なリストを作成します。反復処理を行う場合は、データフレームと一致する電子メールのみをフィルタリングして、Cassandraに保存します。

import sql.implicits._ 
val emailData = sc.parallelize(Seq(("a",1,"abc@gmail.com"), ("b",2,"def@gmail.com"),("a",1,"xyz@gmail.com"),("a",2,"abc@gmail.com"))).toDF("col1","col2","emailId") 
val distinctEmails = emailData.select("emailId").distinct().as[String].collect 
for (email <- distinctEmails){ 
    val subsetEmailsDF = emailData.filter($"emailId" === email).coalesce(1) 
    //... Save the subset dataframe to cassandra 
} 

注意：coalesce（1）はすべてのデータを1つのノードに送信します。データフレームが大きすぎると、メモリの問題が発生することがあります。