I have been running a very large use cases and due to resource constraint, I have to limit the no. of data being processed at a given time. So running a counter for loop such as 1 to 10 and for every run I process, say 200000 rows.
My question is, if my source has 1M rows, I would end up processing for 5 runs (1M/ 200000). And I have multiple sources also. So, for eg.
Source A has 1M rows
Source B has 2M rows.
When i am in the loop for each Source, say,
for (source in Sources) {
val rddSource = spark.read.table(source)
rddSource.persist
rddSource.count // to persist the above rdd
while (counter < 5) {
//process every 200000 rows with the above persisted rddSource
}
}
I want to persist the data as shown above.that will help to run Source A for every 5 time each with 200000 rows.
But for next iteration of for loop with the Source B, does it replace the previously cached {rddSource} or should I need to have something like this
for (source in Sources) {
val rddSource = spark.read.table(source)
rddSource.unPersist
rddSource.persist
rddSource.count // to persist the above rdd
while (counter < 5) {
//process every 200000 rows with the above persisted rddSource
}
}