A catching DList for data-quality

[Eric Springer]

An issue I've been running into a lot, is I might be reading
something, say reading some JSON lines to get the armySize field


```
fromTextFile("foo").map { line =>
JSON.parseFull(line).asInstanceOf[Option[Map[String,
Any]].get.apply("armySize")
}
```

and life is sweet, and runs just fine on my sampled data.


But when I run on a larger dataset, I get an exception -- say: "Can't
call head on empty Option". A while later, I've run a few more
map-reduce jobs and found the offending line -- there's some empty
lines and some map-formed json. So instead of returning getting
`Map[String, Any]` I'll change it to a `Either[Map[String, Any],
MalFormedLineInfo]` and continue on my day.

But oh shit. It's failing again. Wth? Oh, crap -- it can't get a
title? Again, go through all the motions and indeed confirm its a data
problem. Maybe there's a couple of ancient records that predate the
requirement that this map needs an records needs an armySize. So I
repeat the process, etc.


And while I in generally I'd much prefer being explicit (e.g. using
Either's and Options or w/e) -- the thing is, in like the example
above -- I had no idea that it was Option. I totally believed that
every line was a Json Map with an army size, but I had to waste a
massive amount of time, to figure out the real cause.

On possible solution is I could just always write:

fromTextFile("foo").mapFlatten { line =>
JSON.parseFull(line).asInstanceOf[Option[Any]]
}

and just let scoobi filter out everything that doesn't meet my
requirements. But this scares the living crap out of me, as now the
code silently drops everything that doesn't meet my ill-informed
belief. And in reality, I think this will end

So what I would almost like is a "catching DList" (or something) that
effectively has the `process` function of `DoFn` surrounded with a
Try { } block -- and would then keep track of all the Exceptions and
Input that had to be skipped. And ideally after running a job it'd be
like:


BIG ASS WARNING: 2343 RECORDS (0.01%) HAD TO BE DROPPED BECAUSE YOU
SUCK, view details: X


or something :d

[Alex Cozzi]

Actually that is a great idea. Cascading does something similar with failure traps (http://docs.cascading.org/cascading/2.0/userguide/htmlsingle/#N21366) which I found fairly useful.

[Eric Torreborre]

Hi Eric,

Why don't you go with your own library support:

  def parseFile[T](list: DList[String], f: String => T): (DList[T], DList[String]) = {

     val parsed = list.map(s => try Right(f(s)) catch { case e => Left(e.getMessage) })

     val errors = parsed.collect { case Left(e) => e }

     val successes  = parsed.collect { case Right(r) => r }

     (successes, errors)

  }

Then you can use/persist those 2 lists how you wish.

Cheers,

Eric.

[Eric Springer]

Oh, no, definitely. I think that's the proper solution -- even if
there was a "catching DList". But my point was, it took me several
hours to learn that there was things I couldn't parse, to investigate
why, then talk to some people about the data. And then settle on
something like that. And then finally restart my job. And this happens
reasonably often.

But the point of the catching DList would be to allow the job to run
to completion, and then at my leisure start digging into the
exceptions.

That said, it's pretty low on my wish-list of things. My current #1
goes to having a fast in-memory-mode :D [At the moment, inmemory
mode is taking about ~2 minutes, and using >4GB of memory for a
reasonably complex job with 1000 elements. The same thing with scala
collections is running under a second :D

> --
> You received this message because you are subscribed to the Google Groups
> "scoobi-dev" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to scoobi-dev+...@googlegroups.com.
> For more options, visit https://groups.google.com/groups/opt_out.
>
>

[Ben Lever]

Hi EricS,

To be honest we haven't spent any time on benchmarking the performance of in-memory mode :( Under the hood, it just uses Scala Vector objects. What is your pure Scala collections implementation like? If you can, would be great if you could determine where the time is going in Scoobi's in-memory implementation ... the code is also fairly simple and isolated so you may be able to attempt a fix yourself.

Cheers.

[Eric Springer]

My in memory version too just uses a Stream instead of DLists and
values instead of DObjects, so I'd imagine it's pretty much the same.
Although, when I run inmemory mode, I notice the console is getting
super spammed with file io logs, so that's definitely once difference.
I've still got 2 or 3 more bugs I need to minimize for you guys
(including one that computes incorrect values) but after that I'll see
if my inmemory slowness is anything obvious :D