Spark backend #1832
Spark backend #1832
Conversation
|
@ianoc can you take a look? This is not finished, but what we have is testable and in the interest of keeping the PRs small, I'd like to merge this and then follow up with more parts:
|
|
build fails because spark is not there for 2.12. Will remove spark 2.12 from the CI. |
| object SparkPlanner { | ||
|
|
||
| // TODO, this may be just inefficient, or it may be wrong | ||
| implicit private def fakeClassTag[A]: ClassTag[A] = ClassTag(classOf[AnyRef]).asInstanceOf[ClassTag[A]] |
There was a problem hiding this comment.
i think they just register these to kryo so i imagine this should just hit the inefficient paths. For a normal execution app though this i guess does drop some sort of performance.
| case (ForceToDisk(pipe), rec) => | ||
| rec(pipe).persist(StorageLevel.DISK_ONLY) | ||
| case (Fork(pipe), rec) => | ||
| rec(pipe).persist(StorageLevel.MEMORY_ONLY) |
There was a problem hiding this comment.
don't need to bother updating this here, but i'd leave these to DISK_ONLY until we can figure out how/when we can upgrade the state. (Spark also has the output of shuffles persisted, so we might be able to get the planner to realize when we should have shuffle data and not do the persisting then).
| ??? | ||
| case (slk @ SumByLocalKeys(_, _), rec) => | ||
| def sum[K, V](sblk: SumByLocalKeys[K, V]): R[(K, V)] = { | ||
| // we can use Algebird's SummingCache https://github.com/twitter/algebird/blob/develop/algebird-core/src/main/scala/com/twitter/algebird/SummingCache.scala#L36 |
There was a problem hiding this comment.
since these partitions are ~usually ondisk or can fit in memory we might just want MapAlgebra.sumByKey possibly (or sort and fold i guess..).
|
Some comments but they are more of an ongoing discussion than anything else. LGTM |
|
thanks for the comments. these seem like good points to keep in mind as we optimize. All of your comments are doable for sure. Will address in the follow ups. |
This is a very basic beginning to a spark backend.
It is not complete, but does support map-only operations.
There is one big question: can we really just lie to spark and say we have AnyRef everywhere? I think it may just make serialization worse (kryo writing the classnames), but maybe we can circumvent that later since scalding allows configs to have registered classes named, maybe we can pass that information to spark somehow.
cc @ianoc