Spark Streaming - mit foreachPartition und saveToCassandra für bessere Parallelisierung

Question

Unten Code funktioniert gut, aber es dauert einige Zeit, um nach Cassandra zu schreiben, wenn wir einen riesigen Zustrom von Transaktionen haben.

Unten Code ist schriftlich in Cassandra in sequenziellen und führt in einem Executor.

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

Aber ich wollte den Schreibvorgang an Cassandra parallelisieren, indem ich foreachPartition mache. Aber ich sehe nicht saveToCassandra Option bei ForeachPartition.

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

Wie können wir das erreichen?

Answer 1

Da Sie bereits Direct Stream verwenden, gibt es zwei Möglichkeiten, die Parallelität zu erhöhen.

1. Erhöhen Sie die Anzahl der Kafka-Partitionen. Bei der Verwendung von Direct Stream erstellt Spark automatisch so viele Partitionen wie Kafka. Je nach Setup ist dies jedoch möglicherweise nicht möglich.

2. Mit Spark repartition. In den meisten Fällen ist es besser, repartition am Eingang als an der Ausgabe.

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

Wenn Sie nur parsedStream einmal zu verwenden, ist es nicht notwendig, es zu cachen.