scala_api_extensions.md

title	Scala API Extensions
nav-parent_id	api-concepts
nav-pos	10

In order to keep a fair amount of consistency between the Scala and Java APIs, some of the features that allow a high-level of expressiveness in Scala have been left out from the standard APIs for both batch and streaming.

If you want to enjoy the full Scala experience you can choose to opt-in to extensions that enhance the Scala API via implicit conversions.

To use all the available extensions, you can just add a simple import for the DataSet API

{% highlight scala %} import org.apache.flink.api.scala.extensions._ {% endhighlight %}

or the DataStream API

{% highlight scala %} import org.apache.flink.streaming.api.scala.extensions._ {% endhighlight %}

Alternatively, you can import individual extensions a-là-carte to only use those you prefer.

Accept partial functions

Normally, both the DataSet and DataStream APIs don't accept anonymous pattern matching functions to deconstruct tuples, case classes or collections, like the following:

{% highlight scala %} val data: DataSet[(Int, String, Double)] = // [...] data.map { case (id, name, temperature) => // [...] // The previous line causes the following compilation error: // "The argument types of an anonymous function must be fully known. (SLS 8.5)" } {% endhighlight %}

This extension introduces new methods in both the DataSet and DataStream Scala API that have a one-to-one correspondance in the extended API. These delegating methods do support anonymous pattern matching functions.

DataSet API

Method	Original	Example
mapWith	map (DataSet)	{% highlight scala %} data.mapWith { case (_, value) => value.toString } {% endhighlight %}
mapPartitionWith	mapPartition (DataSet)	{% highlight scala %} data.mapPartitionWith { case head #:: _ => head } {% endhighlight %}
flatMapWith	flatMap (DataSet)	{% highlight scala %} data.flatMapWith { case (_, name, visitTimes) => visitTimes.map(name -> _) } {% endhighlight %}
filterWith	filter (DataSet)	{% highlight scala %} data.filterWith { case Train(_, isOnTime) => isOnTime } {% endhighlight %}
reduceWith	reduce (DataSet, GroupedDataSet)	{% highlight scala %} data.reduceWith { case ((_, amount1), (_, amount2)) => amount1 + amount2 } {% endhighlight %}
reduceGroupWith	reduceGroup (GroupedDataSet)	{% highlight scala %} data.reduceGroupWith { case id #:: value #:: _ => id -> value } {% endhighlight %}
groupingBy	groupBy (DataSet)	{% highlight scala %} data.groupingBy { case (id, _, _) => id } {% endhighlight %}
sortGroupWith	sortGroup (GroupedDataSet)	{% highlight scala %} grouped.sortGroupWith(Order.ASCENDING) { case House(_, value) => value } {% endhighlight %}
combineGroupWith	combineGroup (GroupedDataSet)	{% highlight scala %} grouped.combineGroupWith { case header #:: amounts => amounts.sum } {% endhighlight %}
projecting	apply (JoinDataSet, CrossDataSet)	{% highlight scala %} data1.join(data2). whereClause(case (pk, _) => pk). isEqualTo(case (_, fk) => fk). projecting { case ((pk, tx), (products, fk)) => tx -> products } data1.cross(data2).projecting { case ((a, ), (, b) => a -> b } {% endhighlight %}
projecting	apply (CoGroupDataSet)	{% highlight scala %} data1.coGroup(data2). whereClause(case (pk, ) => pk). isEqualTo(case (, fk) => fk). projecting { case (head1 #:: _, head2 #:: _) => head1 -> head2 } } {% endhighlight %}

DataStream API

Method	Original	Example
mapWith	map (DataStream)	{% highlight scala %} data.mapWith { case (_, value) => value.toString } {% endhighlight %}
mapPartitionWith	mapPartition (DataStream)	{% highlight scala %} data.mapPartitionWith { case head #:: _ => head } {% endhighlight %}
flatMapWith	flatMap (DataStream)	{% highlight scala %} data.flatMapWith { case (_, name, visits) => visits.map(name -> _) } {% endhighlight %}
filterWith	filter (DataStream)	{% highlight scala %} data.filterWith { case Train(_, isOnTime) => isOnTime } {% endhighlight %}
keyingBy	keyBy (DataStream)	{% highlight scala %} data.keyingBy { case (id, _, _) => id } {% endhighlight %}
mapWith	map (ConnectedDataStream)	{% highlight scala %} data.mapWith( map1 = case (_, value) => value.toString, map2 = case (_, _, value, _) => value + 1 ) {% endhighlight %}
flatMapWith	flatMap (ConnectedDataStream)	{% highlight scala %} data.flatMapWith( flatMap1 = case (_, json) => parse(json), flatMap2 = case (_, _, json, _) => parse(json) ) {% endhighlight %}
keyingBy	keyBy (ConnectedDataStream)	{% highlight scala %} data.keyingBy( key1 = case (_, timestamp) => timestamp, key2 = case (id, _, _) => id ) {% endhighlight %}
reduceWith	reduce (KeyedDataStream, WindowedDataStream)	{% highlight scala %} data.reduceWith { case ((_, sum1), (_, sum2) => sum1 + sum2 } {% endhighlight %}
foldWith	fold (KeyedDataStream, WindowedDataStream)	{% highlight scala %} data.foldWith(User(bought = 0)) { case (User(b), (_, items)) => User(b + items.size) } {% endhighlight %}
applyWith	apply (WindowedDataStream)	{% highlight scala %} data.applyWith(0)( foldFunction = case (sum, amount) => sum + amount windowFunction = case (k, w, sum) => // [...] ) {% endhighlight %}
projecting	apply (JoinedDataStream)	{% highlight scala %} data1.join(data2). whereClause(case (pk, _) => pk). isEqualTo(case (_, fk) => fk). projecting { case ((pk, tx), (products, fk)) => tx -> products } {% endhighlight %}

For more information on the semantics of each method, please refer to the [DataSet]({{ site.baseurl }}/dev/batch/index.html) and [DataStream]({{ site.baseurl }}/dev/datastream_api.html) API documentation.

To use this extension exclusively, you can add the following import:

{% highlight scala %} import org.apache.flink.api.scala.extensions.acceptPartialFunctions {% endhighlight %}

for the DataSet extensions and

{% highlight scala %} import org.apache.flink.streaming.api.scala.extensions.acceptPartialFunctions {% endhighlight %}

The following snippet shows a minimal example of how to use these extension methods together (with the DataSet API):

{% highlight scala %} object Main { import org.apache.flink.api.scala.extensions._ case class Point(x: Double, y: Double) def main(args: Array[String]): Unit = { val env = ExecutionEnvironment.getExecutionEnvironment val ds = env.fromElements(Point(1, 2), Point(3, 4), Point(5, 6)) ds.filterWith { case Point(x, _) => x > 1 }.reduceWith { case (Point(x1, y1), (Point(x2, y2))) => Point(x1 + y1, x2 + y2) }.mapWith { case Point(x, y) => (x, y) }.flatMapWith { case (x, y) => Seq("x" -> x, "y" -> y) }.groupingBy { case (id, value) => id } } } {% endhighlight %}