シェアsparkRDD使用して、私が実装何のapacheのIgnite

Question

は次のとおりです。

val sparkConf = new SparkConf().setAppName(“SharedRDD”).setMaster(“local”) 
val sc = SparkContext.getOrCreate(sparkConf) 
val sparkRDD = sc.wholeTextFiles("sample.csv", 10) 

これRDDは、今、今後

val igniteContext = new IgniteContext(sc, “example-shared-rdd.xml”, false) 

val sharedIgniteRDD = igniteContext.fromCache[String, String](“cachedIgniteRDD”) 
if (sharedIgniteRDD.isEmpty()) 
sharedIgniteRDD.savePairs(sparkRDD) 

IgniteContext

によってキャッシュされている任意のスパーク仕事をこのRDDにアクセスする必要がある場合、それが作成する必要はありません代わりに新しいものが点火キャッシュからそれを取り出す。入力ファイルの

val RDDfromCache = igniteContext.fromCache[String, String](“CachedIgniteRDD”) 

サンプルデータ

25/07/13,11599,CLOSED 
25/07/13,256,PENDING_PAYMENT 
25/07/13,12111,COMPLETE 
25/07/13,8827,CLOSED 
25/07/13,11318,COMPLETE 
25/07/13,7130,COMPLETE 
25/07/13,4530,COMPLETE 
25/07/13,2911,PROCESSING 
25/07/13,5657,PENDING_PAYMENT 
25/07/13,5648,PENDING_PAYMENT 

私はその入力ファイルのステータス列にグループ分けを行うにgroupByKey（）APIを呼び出す必要があります。

あなたのご協力をお待ちしております。 cacheまたはpersistの

おかげ

Answer 1

使用すると、再び再作成するrddを得ることから避けてきたはずです。そして、保存するオプションがあります。rddsin-memory、memory-disk、serialized、deserializedなどIgniteContextは上記の解決方法の代替品です。

groupByKeyの保存にはrddが少しありますが、wholeTextFileを使ってテキストファイルを読むと、IgniteContextになります。 WholeTextFileあなたがIgniteContext

val RDDfromCache = igniteContext.fromCache[String, String](“CachedIgniteRDD”) 

の保存rddfromCacheを読んでたら、あなたはgroupByKeyために次の操作を行うことができますTuple2のRDD（ "テキストファイルへのパス"、 "テキスト行 S"）

を生成します。

val groupedRDD = RDDfromCache.flatMap(x => x._2.split("\n")).map(array => (array.split(",")(2), array)).groupByKey().foreach(println) 

出力は次のようになり

(CLOSED,CompactBuffer(25/07/13,11599,CLOSED, 25/07/13,8827,CLOSED)) 
(PENDING_PAYMENT,CompactBuffer(25/07/13,256,PENDING_PAYMENT, 25/07/13,5657,PENDING_PAYMENT, 25/07/13,5648,PENDING_PAYMENT)) 
............ 
.......... 
.......... 

Answer 2

wholeTextFilesは、ファイル全体を1つのエントリに入れているため、Igniteに保存することはあまり意味がありません。基本的には1つの大きなエントリを持つ分散キャッシュがあります。これは利点を提供しません。

ファイルを分割して、各行を別々のタプルとして保存する必要があります。このデータをIgniteに保存し、RDD APIを使用して処理することができます。