Dataset design

[camjo]

Creating this issue to discuss the detailed design of Dataset[A] (and potentially DataStream[A]).

As discussed in the meeting with John, I think its worth thinking through what this API would look like as both batch and stream. We can go severals ways with this:

The first would be a unified API like Spark's Dataset (and structured streaming).
https://spark.apache.org/docs/2.3.1/structured-streaming-programming-guide.html#programming-model

Another would be Flink's Dataset/DataStream API (which are build on top of their stateful streams abstraction).
https://ci.apache.org/projects/flink/flink-docs-release-1.5/concepts/programming-model.html

I'd like to flesh out what this API should look like and how it should function in more detail here.

[SemanticBeeng]

Do we agree this cannot be separated from the data schema discussion?

In the above I guess that A would be the data schema at a logical level but given that apache arrow is in scope (great!) one can conclude we need to consider implementation reuse and try to "lift" them into design.

Concretely, apache spark, apache arrow, avro and parquet all have elements of data schema.

Then apache arrow also implies interoperability with Python, C++ and even R are in scope (this should).
Are they?

If so then we can consider data schema support from numpy, xarray , xtensor, etc.

Lastly we have jep and its ability to "zero copy" share memory between JVM and CPython runtime.

So I think we need to include the above in scope - thoughts please?

[camjo]

@SemanticBeeng If I'm understanding you correctly, I think the data schema you're referring to can be separated from the API design discussion. What we are aiming to do here is define an API that describes the operations we want to support. The actual data formats in memory or on disk are a separate concern and will occur when we interpret the Dataset DAG to some back end (at which point this will become an important discussion).

[diminou]

Having a unified API for both streaming and batch use cases means forbidding several useful operations on batch datasets (computing count and other basic aggregates come to mind; Spark forbids some), as they are meaningless for streams. However, those same operations are very practical for quick explorations of batches from the REPL.
It could be useful to have a streaming API by default with extensions for the case where the data are not a stream, so that it the users doing batch-mode analysis may benefit from the ergonomic standards other tools provide.
So, a Dataset[A] may be manipulated in any way a DataStream[A] may, but being tailored for batch data, it would have additional simplified operations, like count returning a number instead of a stream.

[camjo]

Yeah so ultimately, the low level description language probably makes sense to be streaming based since streams are a superset of batch operations: https://data-artisans.com/blog/batch-is-a-special-case-of-streaming

Ultimately the user facing APIs will just be a way to construct this lower level description of a program. I think that exposing two APIs (Dataset[A] and DataStream[A]) might be the right way to go. My thinking on why is:

• Both of these can easily translate down to the actual grammar we want to interpret and manipulate.
• Whilst batch is technically a special case of streaming, its still much easier to program batch logic without worrying about streaming concepts. This also means the APIs will be more streamlined for purpose rather than being cluttered with windows etc when all you're trying to do is read and count * on a file.
• We can translate between the two API if/when it makes sense (not unlike the KStreams API for example which exposes a KTable and a KStream and lets you translate between them - though we need to consider the semantics of this translation because they treat a KTable as a point in time aggregation of a KStream (i.e. their "batch" api is still technically a streaming api. You can't make assumptions that the KTable is "fully loaded" etc))
• We still have the ability (like flink) to layer higher level abstractions on top such as the "Table API" which unifies the batch and streaming models if thats what people want by constraining the set of things that can be done (similar as you mentioned to spark).

Thoughts?

[diminou]

@camjo absolutely, that would buy us the ability to implement operations over Dataset separately from DataStream and without thinking about streaming implications. And when both are sufficiently complete, a rewrite of some of the Dataset's API in terms of DataStream's could help us reduce duplication.
In the meantime, if needed, an umbrella (such as flink's Table API you mentioned) would aggregate minimal common API operations.

[camjo]

I have a very rough API design proposal here: https://gist.github.com/camjo/10cb0f25b9da10f08f9b30cbd9419985

I've done a lot of thinking about the nature of batch and stream and what it means to do computation on each as well as how we should expose it as an API.

I think it fundamentally boils down to data vs. computation. Data can roughly be conceptualised as either Bounded or Unbounded. Computation can be thought of as (online/streaming/unbounded) vs. (offline/batch/bounded). The Gist explains what I mean by this in more detail. Hopefully you can follow the types to understand how the various concepts fit together.

There are still a fair few gaps listed as // todo throughout the gist.

@jdegoes Would be keen to get your thoughts on this so far. Would it be easier as a PR that we can work on improving before merging or is a gist fine for now?

Perhaps if this seems like a suitable direction, it can form the basis of another design session. I'm aware that its not implemented in terms of =>: etc yet but I can clean that up shortly if the general structure looks correct.

[jdegoes]

@camjo Another session works, too.

Some quick thoughts:

1. We don't need GroupedDataset, we should use Dataset[Map[K, V]] for this, or perhaps, Dataset[Grouped[K, V]], where Grouped is our own type. Then we can define a family of aggregations Grouped[K, V] =>: V2 (we have to constrain aggregations some way rather than take in Scala closures, but we can still be very flexible if necessary).
2. Similarly, I don't think we need WindowedDataStream, I think we can use DataStream[Grouped[K, V]].
3. In general, move away from the Scala closures. We may have to do some design work here.
4. Overall the methods between Dataset and DataStream look so similar, I'm tempted to unify them. Then we just have something like Dataset[Window[A, B]] for a windowed view, and some operations only work on that.

[camjo]

@jdegoes

1. Hmm thats an interesting idea. I initially thought "it should be Grouped[K, List[V]] but perhaps you're thinking of a more general Aggregatable structure for V that can be used by implementing modules?

2. I'm not sure that this fully captures all the possible window semantics. e.g. how do we represent Sliding vs. rolling vs. hopping windows if we only have the equivalent of a Map[K, V]? (See below for why I think we need at least 3 distinct cases - though I think we can potentially collapse GroupedDataset as you suggested). Either way it seems like we would need to treat DataStream[Grouped[K, V]] as if it were a separate case anyway. Is this just renaming types rather than removing the concept of WindowedDataStream?

3. Agree - this is just a quick and dirty impl to work out the heirachy of ops. Will update this shortly.

4. Yeah so I spent quite a while thinking through this lately. My first prototype actually unified them but it's really challenging to express the nature and computations of both batches and streams using only streams. I agree its a tempting goal that we should explore but I actually think you lose the ability to talk fully about one vs. the other in detail if you do that. Happy to be proven wrong here.

The issue seems to be that you have 3 cases caused by the data/computation split.

Data   |   Computation
-------------------------------------------------------------
batch  |   bounded (map/filter/fold/distinct etc)
stream |   bounded (windowed, e.g. fold/distinct/average etc)
stream |   unbounded (e.g. map/filter etc)

And consequently

• A bounded computation on a batch source is always a batch output.
• A bounded computation on a stream (i.e. windowed) actually returns another stream (think sliding sum as an example)
• An unbounded computation on a stream also returns another stream.

Note specifically that a bounded computation on a batch doesn't seem to have the same signature as a bounded computation on a stream.

The alternative here would be to say "ok lets treat a batch as a finite stream and collapse them". Ok sure, but you're not removing any of those cases, you're simply renaming "batch" to "finite stream". The 3 cases of semantics still need to exist. Perhaps there is a more precise encoding that I'm missing though?

In terms of prior work:

• Spark added streaming semantics to their Dataset API which we could take a look at in terms of its implementation/limitations.
• Kafka Streams treats everything as a stream but doesn't give a language for bounded streams (at least that I'm aware of) which makes it hard to enforce the kinds of semantics we want in an analytics engine.
• Flink has two APIs that sit on top of a Structured Streaming core - which is an arbitrary processing node that can choose how many messages it processes at a time (effectively allowing it to emulate batch processing - though I'm not yet clear how they mark that a batch is all there. This could be where to look for ideas on how to express a batch in a streaming system).
• Beam treats everything as either a stream or a table. I extracted the essence of the heirachy in this gist: https://gist.github.com/camjo/75ca375d6569b790024016b75d3f6e90.

Thoughts?

[idc101]

r.e. GroupedDataset, from a usability point of view I found Spark's equivalent hard to work with and usually dropped back to Spark's un-typed DataFrames for group by/aggregations. It would be interesting to explore Dataset[Grouped[K, List[V]]] further to see how usable it would be.

[camjo]

@idc101
Just thinking about it, what is the purpose of grouping a dataset without aggregating it? Do we gain anything special by exposing it as a new state? Perhaps we can just remove that transition entirely by making every groupby have an aggregate at construction time? AFAIK SQL doesn't allow you to group by without an aggregation of some sort either.

It seems to me that the usage of a group by in frameworks that expose it with an intermediate structure like GroupedDataset falls into one of two buckets (please tell me if I’ve missed something here):

1. A fancy DSL for adding aggregations (e.g. gds.min('col1).max('col4)
2. Getting a reference to the grouped dataset to use it as a new “source” for a set of branching aggregations.
   e.g.

val dataset = ???
val grouped = ds.groupBy(a => a.b)
val output1 = grouped.agg(min, max)
val output2 = grouped.agg(sum, count)
// use output1 and output2 separately

The thing is, other libraries/frameworks can't express it like this:

val dataset = ???
val output1 = ds.groupBy(a => a.b, min, max)
val output2 = ds.groupBy(a => a.b, sum, count)

Reason being is that they allow arbitrary functions in the group by clause so they can't optimise common aggregates. Since we capture functions and reify everything we can actually provide that API and still optimise it to run more like the first code snippit above by comparing equality of the functions.

This would remove that state entirely.

Can anyone see why this might not work?

I wonder if we could even apply the same logic to WindowedDataStream as well? I haven't thought that through properly yet.

[Umko]

I like the idea of data vs computation split, since the nature of computation on a stream is often very different from the one on a batch resulting in different optimisation strategies which might be applied and different expectations to a runtime.

AFAIK SQL doesn't allow you to group by without an aggregation of some sort either.

you can - if you group by all columns, result will be an equivalent of a distinct (surprisingly producing more efficient query plan in some cases)