Hi,
Can anyone please help clarity on how accumulators can be used reliably to
measure error/success/analytical metrics ?
Given below is use case / code snippet that I have.
val amtZero = sc.accumulator(0)
> val amtLarge = sc.accumulator(0)
> val amtNormal = sc.accumulator(0)
> val getAmount = (x: org.apache.spark.sql.Row) => if (x.isNullAt(somePos)) {
> amtZero.add(1)
> 0.0
> } else {
> val amount = x.getDouble(4)
> if (amount > 10000) amtLarge.add(1) else amtNormal.add(1)
> amount
> }
> mrdd = rdd.map(s => (s, getAmount(s)))
> mrdd.cache()
> another_mrdd = rdd2.map(s => (s, getAmount(s)))
> mrdd.save_to_redshift
> another_mrdd.save_to_redshift
> mrdd.union(another_mrdd).map().groupByKey().save_to_redshift
// Get values from accumulators and persist it to a reliable store for
> analytics.
> save_to_datastore(amtZero.value, amtLarge.value, amtNormal.value)
Few questions :
1. How many times should I expect the counts for items within mrdd and
another_mrdd since both of these rdd's are being reused ? What happens when
a part of DAG is skipped due to caching in between (say I'm caching
only mrdd)?
2. Should I be worried about any possible stage/task failures (due to
master-wroker network issues/resource-starvation/speculative-execution),
can these events lead to wrong counts ?
3. Document says **In transformations, users should be aware of that each
task’s update may be applied more than once if tasks or job stages are
re-executed.**
Here re-execution of stages/tasks is referring to failure re-executions or
re-execution due to stage/tasks position in DAG ?
4. Is it safe to say that usage of accumulators(for exact counts) are
limited to .foreach() as actions guarantee exactly once updates ?
Thanks
Sudev
