sparkストリーミング - より良い並列化のためにforeachPartitionとsaveToCassandraを使用

Question

以下のコードはうまくいきますが、トランザクションが大量に流入すると、cassandraに書き込むのに時間がかかります。

val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics) 
val parsedStream = stream.map(_._2).map(EmpParser.parse(_)).cache() 

以下のコードは、cassandraに順次書き込みを行い、1つのエグゼキュータで実行します。

parsedStream.saveToCassandra("test", "ct_table", SomeColumns("emp_id","emp_name","emp_sal","emp_dept")) 

しかし、foreachPartitionを実行して、cassandraへの書き込みを並列化したかったのです。しかし、foreachPartitionでsaveToCassandraオプションが表示されません。

parsedStream.foreachRDD{rdd => 
    rdd.foreachPartition { partition => 
     partition.saveToCassandra("test", "ct_table", SomeColumns("emp_id","emp_name","emp_sal","emp_dept")) 
    } 
} 

これを実現する方法はありますか。

Answer 1

すでにダイレクトストリームを使用しているため、並列性を高める方法は2つあります。

1. カフカパーティションの数を増やします。ダイレクトストリームを使用する場合、Sparkは自動的にKafkaと同じ数のパーティションを作成します。ただし、セットアップによっては、これが実現できない場合があります。

2. スパークスrepartitionを使用してください。ほとんどの場合、出力ではなく入力上でrepartitionが良いです。

   val num: Int = ? // Number of partitition 
   val parsedStream = stream.repartition(num).map(_._2).map(EmpParser.parse(_)) 
   parsedStream.saveToCassandra(...) 
   

あなたは一度だけparsedStreamを使用する必要がある場合は、それをキャッシュする必要はありません。