doing cleanup in an iterator like that assumes the iterator always gets
fully read, which is not necessary the case (for example RDD.take does not).
instead i would use mapPartitionsWithContext, in which case you can write a
function of the form.
f: (TaskContext, Iterator[T]) => Iterator[U]
now you can register a cleanup with the task context, like this:
context.addTaskCompletionListener(new TaskCompletionListener {
override def onTaskCompletion(context: TaskContext): Unit = dosomething
)
and after that proceed with an iterator transformation as usual
On Thu, Jul 31, 2014 at 4:35 AM, Sean Owen <so...@cloudera.com> wrote:
> On Thu, Jul 31, 2014 at 2:11 AM, Tobias Pfeiffer <t...@preferred.jp> wrote:
> > rdd.mapPartitions { partition =>
> > // Some setup code here
> > val result = partition.map(yourfunction)
> >
> > // Some cleanup code here
> > result
> > }
>
> Yes, I realized that after I hit send. You definitely have to store
> and return the result from the mapping!
>
>
> > rdd.mapPartitions { partition =>
> > if (!partition.isEmpty) {
> >
> > // Some setup code here
> > partition.map(item => {
> > val output = yourfunction(item)
> > if (!partition.hasNext) {
> > // Some cleanup code here
> > }
> > output
> > })
> > } else {
> > // return an empty Iterator of your return type
> > }
> > }
>
> Great point, yeah. If you knew the number of values were small you
> could collect them and process locally, but this is the right general
> way to do it.
>
