You can also use short-hand syntax for minimum, maximum, and sum aggregations.
{% highlight scala %} @@ -1037,7 +1037,7 @@ val csvInput = env.readCsvFile[Person]( val values = env.fromElements("Foo", "bar", "foobar", "fubar") // generate a number sequence -val numbers = env.generateSequence(1, 10000000); +val numbers = env.generateSequence(1, 10000000) // read a file from the specified path of type TextInputFormat val tuples = env.readHadoopFile(new TextInputFormat, classOf[LongWritable], @@ -1288,7 +1288,7 @@ val values: DataSet[(String, Int, Double)] = // [...] values.writeAsCsv("file:///path/to/the/result/file", "\n", "|") // this writes tuples in the text formatting "(a, b, c)", rather than as CSV lines -values.writeAsText("file:///path/to/the/result/file"); +values.writeAsText("file:///path/to/the/result/file") // this writes values as strings using a user-defined formatting values map { tuple => tuple._1 + " - " + tuple._2 } @@ -1309,19 +1309,19 @@ val pData: DataSet[(BookPojo, Double)] = // [...] val sData: DataSet[String] = // [...] // sort output on String field in ascending order -tData.sortPartition(1, Order.ASCENDING).print; +tData.sortPartition(1, Order.ASCENDING).print() // sort output on Double field in descending and Int field in ascending order -tData.sortPartition(2, Order.DESCENDING).sortPartition(0, Order.ASCENDING).print; +tData.sortPartition(2, Order.DESCENDING).sortPartition(0, Order.ASCENDING).print() // sort output on the "author" field of nested BookPojo in descending order -pData.sortPartition("_1.author", Order.DESCENDING).writeAsText(...); +pData.sortPartition("_1.author", Order.DESCENDING).writeAsText(...) // sort output on the full tuple in ascending order -tData.sortPartition("_", Order.ASCENDING).writeAsCsv(...); +tData.sortPartition("_", Order.ASCENDING).writeAsCsv(...) // sort atomic type (String) output in descending order -sData.sortPartition("_", Order.DESCENDING).writeAsText(...); +sData.sortPartition("_", Order.DESCENDING).writeAsText(...) {% endhighlight %} @@ -1486,7 +1486,7 @@ val result = count map { c => c / 10000.0 * 4 } result.print() -env.execute("Iterative Pi Example"); +env.execute("Iterative Pi Example") {% endhighlight %} You can also check out the @@ -1693,7 +1693,7 @@ val env = ExecutionEnvironment.createLocalEnvironment() val lines = env.readTextFile(pathToTextFile) // build your program -env.execute(); +env.execute() {% endhighlight %} diff --git a/docs/dev/connectors/elasticsearch.md b/docs/dev/connectors/elasticsearch.md index b6ee63c0449f8a002b1ab9e7a26d19ed9bdb14da..8774fcbcee868e8d1fbe0f8adcda88c57a813237 100644 --- a/docs/dev/connectors/elasticsearch.md +++ b/docs/dev/connectors/elasticsearch.md @@ -159,7 +159,7 @@ input.addSink(new ElasticsearchSink(config, transportAddresses, new Elasticsearc return Requests.indexRequest() .index("my-index") .type("my-type") - .source(json); + .source(json) } })) {% endhighlight %} @@ -185,7 +185,7 @@ input.addSink(new ElasticsearchSink(config, transportAddresses, new Elasticsearc return Requests.indexRequest() .index("my-index") .type("my-type") - .source(json); + .source(json) } })) {% endhighlight %} @@ -298,7 +298,7 @@ input.addSink(new ElasticsearchSink(config, new ElasticsearchSinkFunction[String return Requests.indexRequest() .index("my-index") .type("my-type") - .source(json); + .source(json) } })) {% endhighlight %} diff --git a/docs/dev/connectors/kafka.md b/docs/dev/connectors/kafka.md index 6c80370db4089cdff19a5f8b8d21437d4d61d4c9..daf1903502ffadf7be42f84f5992b815be79bcfd 100644 --- a/docs/dev/connectors/kafka.md +++ b/docs/dev/connectors/kafka.md @@ -130,14 +130,14 @@ DataStreamBy default a relaxed internal contiguity (between subsequent events) is used. For more info on internal contiguity see consecutive.
{% highlight scala %} -pattern.times(2, 4); +pattern.times(2, 4) {% endhighlight %} @@ -765,7 +765,7 @@ pattern.oneOrMore().greedy() Pattern.begin("start").where(_.getName().equals("c")) .followedBy("middle").where(_.getName().equals("a")) .oneOrMore().consecutive() - .followedBy("end1").where(_.getName().equals("b")); + .followedBy("end1").where(_.getName().equals("b")) {% endhighlight %}Will generate the following matches for an input sequence: C D A1 A2 A3 D A4 B
@@ -786,7 +786,7 @@ Pattern.begin("start").where(_.getName().equals("c")) Pattern.begin("start").where(_.getName().equals("c")) .followedBy("middle").where(_.getName().equals("a")) .oneOrMore().allowCombinations() - .followedBy("end1").where(_.getName().equals("b")); + .followedBy("end1").where(_.getName().equals("b")) {% endhighlight %}Will generate the following matches for an input sequence: C D A1 A2 A3 D A4 B
@@ -1491,7 +1491,7 @@ val result: SingleOutputStreamOperator[ComplexEvent] = patternStream.select(outp pattern: Map[String, Iterable[Event]] => ComplexEvent() } -val timeoutResult: DataStreamSimilar to a SQL OVER clause. Over window aggregates are computed for each row, based on a window (range) of preceding and succeeding rows. See the over windows section for more details.
- {% highlight scala %} +{% highlight java %} Table orders = tableEnv.scan("Orders"); Table result = orders // define window @@ -364,8 +364,8 @@ Table result = orders .orderBy("rowtime") .preceding("UNBOUNDED_RANGE") .following("CURRENT_RANGE") - .as("w") - .select("a, b.avg over w, b.max over w, b.min over w") // sliding aggregate + .as("w")) + .select("a, b.avg over w, b.max over w, b.min over w"); // sliding aggregate {% endhighlight %}Note: All aggregates must be defined over the same window, i.e., same partitioning, sorting, and range. Currently, only windows with PRECEDING (UNBOUNDED and bounded) to CURRENT ROW range are supported. Ranges with FOLLOWING are not supported yet. ORDER BY must be specified on a single time attribute.
Note: All aggregates must be defined over the same window, i.e., same partitioning, sorting, and range. Currently, only windows with PRECEDING (UNBOUNDED and bounded) to CURRENT ROW range are supported. Ranges with FOLLOWING are not supported yet. ORDER BY must be specified on a single time attribute.
@@ -614,9 +614,9 @@ Table result = ordersSimilar to a SQL JOIN clause. Joins two tables. Both tables must have distinct field names and at least one equality join predicate must be defined through join operator or using a where or filter operator.
{% highlight scala %} -val left = ds1.toTable(tableEnv, 'a, 'b, 'c); -val right = ds2.toTable(tableEnv, 'd, 'e, 'f); -val result = left.join(right).where('a === 'd).select('a, 'b, 'e); +val left = ds1.toTable(tableEnv, 'a, 'b, 'c) +val right = ds2.toTable(tableEnv, 'd, 'e, 'f) +val result = left.join(right).where('a === 'd).select('a, 'b, 'e) {% endhighlight %}Note: Currently, only INNER time-windowed joins are supported.
Similar to a SQL UNION clause. Unions two tables with duplicate records removed, both tables must have identical field types.
{% highlight scala %} -val left = ds1.toTable(tableEnv, 'a, 'b, 'c); -val right = ds2.toTable(tableEnv, 'a, 'b, 'c); -val result = left.union(right); +val left = ds1.toTable(tableEnv, 'a, 'b, 'c) +val right = ds2.toTable(tableEnv, 'a, 'b, 'c) +val result = left.union(right) {% endhighlight %}Similar to a SQL UNION ALL clause. Unions two tables, both tables must have identical field types.
{% highlight scala %} -val left = ds1.toTable(tableEnv, 'a, 'b, 'c); -val right = ds2.toTable(tableEnv, 'a, 'b, 'c); -val result = left.unionAll(right); +val left = ds1.toTable(tableEnv, 'a, 'b, 'c) +val right = ds2.toTable(tableEnv, 'a, 'b, 'c) +val result = left.unionAll(right) {% endhighlight %}Similar to a SQL INTERSECT clause. Intersect returns records that exist in both tables. If a record is present in one or both tables more than once, it is returned just once, i.e., the resulting table has no duplicate records. Both tables must have identical field types.
{% highlight scala %} -val left = ds1.toTable(tableEnv, 'a, 'b, 'c); -val right = ds2.toTable(tableEnv, 'e, 'f, 'g); -val result = left.intersect(right); +val left = ds1.toTable(tableEnv, 'a, 'b, 'c) +val right = ds2.toTable(tableEnv, 'e, 'f, 'g) +val result = left.intersect(right) {% endhighlight %}Similar to a SQL INTERSECT ALL clause. IntersectAll returns records that exist in both tables. If a record is present in both tables more than once, it is returned as many times as it is present in both tables, i.e., the resulting table might have duplicate records. Both tables must have identical field types.
{% highlight scala %} -val left = ds1.toTable(tableEnv, 'a, 'b, 'c); -val right = ds2.toTable(tableEnv, 'e, 'f, 'g); -val result = left.intersectAll(right); +val left = ds1.toTable(tableEnv, 'a, 'b, 'c) +val right = ds2.toTable(tableEnv, 'e, 'f, 'g) +val result = left.intersectAll(right) {% endhighlight %}Similar to a SQL EXCEPT clause. Minus returns records from the left table that do not exist in the right table. Duplicate records in the left table are returned exactly once, i.e., duplicates are removed. Both tables must have identical field types.
{% highlight scala %} -val left = ds1.toTable(tableEnv, 'a, 'b, 'c); -val right = ds2.toTable(tableEnv, 'a, 'b, 'c); -val result = left.minus(right); +val left = ds1.toTable(tableEnv, 'a, 'b, 'c) +val right = ds2.toTable(tableEnv, 'a, 'b, 'c) +val result = left.minus(right) {% endhighlight %}Similar to a SQL EXCEPT ALL clause. MinusAll returns the records that do not exist in the right table. A record that is present n times in the left table and m times in the right table is returned (n - m) times, i.e., as many duplicates as are present in the right table are removed. Both tables must have identical field types.
{% highlight scala %} -val left = ds1.toTable(tableEnv, 'a, 'b, 'c); -val right = ds2.toTable(tableEnv, 'a, 'b, 'c); -val result = left.minusAll(right); +val left = ds1.toTable(tableEnv, 'a, 'b, 'c) +val right = ds2.toTable(tableEnv, 'a, 'b, 'c) +val result = left.minusAll(right) {% endhighlight %}Similar to a SQL IN clause. In returns true if an expression exists in a given table sub-query. The sub-query table must consist of one column. This column must have the same data type as the expression.
{% highlight scala %} -val left = ds1.toTable(tableEnv, 'a, 'b, 'c); -val right = ds2.toTable(tableEnv, 'a); -val result = left.select('a, 'b, 'c).where('a.in(right)); +val left = ds1.toTable(tableEnv, 'a, 'b, 'c) +val right = ds2.toTable(tableEnv, 'a) +val result = left.select('a, 'b, 'c).where('a.in(right)) {% endhighlight %}Similar to a SQL ORDER BY clause. Returns records globally sorted across all parallel partitions.
{% highlight scala %} -val in = ds.toTable(tableEnv, 'a, 'b, 'c); -val result = in.orderBy('a.asc); +val in = ds.toTable(tableEnv, 'a, 'b, 'c) +val result = in.orderBy('a.asc) {% endhighlight %}