If else in Spark Streaming

Question

I have a Spark Streaming application which is reading data from a SINGLE TOPIC in Kafka, processing it, and inserting it into 2 different keyspaces in Cassandra, based on the content of the element. Some data might go to keyspace A, some other to keyspace B.

I do it currently using filter operation:

Functions.insertToCassandra(rdd.filter(element => element.tenant=="A"), keyspace = A, table = "tableName")
Functions.insertToCassandra(rdd.filter(element => element.tenant=="B"), keyspace = B, table = "tableName")

So filter is applied on every rdd, those elements that have tenant field A go to keyspace A, and those with tenant field B go to keyspace B.

Is there any more efficient way to do this, instead of applying filter operation 2 times (especially because there might be more than 2 keyspaces later)? Would caching rdd before filter operations increase the performance?

I repeat, I have DStream coming from Kafka, I process it, and then in "foreachRDD" operation I have code snippet from above, which is inserting data to Cassandra.

Thank you

Answer 1

Before you do

Functions.insertToCassandra(rdd.filter(element => element.tenant=="A"), keyspace = A, "tableName")
Functions.insertToCassandra(rdd.filter(element=> element.tenant=="B"), keyspace = B, "tableName")

Make sure to do rdd.cache()

When you are doing like above, your spark is trying to read rdd data twice. Spark never retains any rdd in memory unless you cache or broadcast it.

Another way could be to read all the data at once, cache it, if data set is not enormous. Then use groupByKey, where key will be your keyspace(element) in this case.