Spark Streaming - использование 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 в последовательном порядке и выполняется в одном исполнителе.

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

Но я хотел распараллелить запись в cassandra, выполнив foreachPartition. Но я не вижу вариант saveToCassandra в foreachPartition.

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

Любой способ добиться этого?

Answer 1

Поскольку вы уже используете прямой поток, есть два способа увеличить параллелизм.

1. Увеличить количество разделов Kafka. При использовании прямого потока Spark автоматически создает столько разделов, как Kafka. Однако, в зависимости от вашей настройки, это может оказаться невыполнимым.

2. Использование Spark's repartition. В большинстве случаев лучше использовать repartition на входе, а не на выходе.

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

Если вам нужно только использовать parsedStream один раз, это не нужно кэшировать.