You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2019/05/13 09:43:37 UTC
[flink] branch master updated: [FLINK-12421][docs-zh] Synchronize
the latest documentation changes (commits to 754cd71d) into Chinese
documents
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3f532e1 [FLINK-12421][docs-zh] Synchronize the latest documentation changes (commits to 754cd71d) into Chinese documents
3f532e1 is described below
commit 3f532e18b96b83abdde189c4304f66c60b285d5c
Author: Jark Wu <im...@gmail.com>
AuthorDate: Tue May 7 10:11:08 2019 +0800
[FLINK-12421][docs-zh] Synchronize the latest documentation changes (commits to 754cd71d) into Chinese documents
This closes #8354
---
docs/_includes/sidenav.html | 5 +-
docs/dev/api_concepts.zh.md | 2 +-
docs/dev/connectors/kafka.md | 2 +-
docs/dev/connectors/kafka.zh.md | 2 +-
docs/dev/execution.zh.md | 2 +-
docs/dev/execution_configuration.zh.md | 2 +-
docs/dev/execution_plans.zh.md | 2 +-
docs/dev/packaging.zh.md | 4 +-
docs/dev/parallel.zh.md | 2 +-
docs/dev/restart_strategies.zh.md | 2 +-
docs/dev/stream/state/queryable_state.zh.md | 44 +--
docs/dev/table/functions.zh.md | 402 +++++++++++++++++++++++-----
docs/dev/table/tableApi.zh.md | 294 +++++++++++++++++++-
docs/ops/deployment/yarn_setup.zh.md | 1 +
docs/ops/index.zh.md | 2 +-
15 files changed, 673 insertions(+), 95 deletions(-)
diff --git a/docs/_includes/sidenav.html b/docs/_includes/sidenav.html
index 247a0e3..7f200b1 100644
--- a/docs/_includes/sidenav.html
+++ b/docs/_includes/sidenav.html
@@ -116,7 +116,10 @@ level is determined by 'nav-pos'.
{% capture collapse_target %}"#collapse-{{ i }}" data-toggle="collapse"{% if active %} class="active"{% endif %}{% endcapture %}
{% capture expand %}{% unless active %} <i class="fa fa-caret-down pull-right" aria-hidden="true" style="padding-top: 4px"></i>{% endunless %}{% endcapture %}
<li><a href={{ collapse_target }}>{{ title }}{{ expand }}</a><div class="collapse{% if active %} in{% endif %}" id="collapse-{{ i }}"><ul>
- {% if this.nav-show_overview %}<li><a href={{ overview_target }}>Overview</a></li>{% endif %}
+ {% if this.nav-show_overview %}
+ <li><a href={{ overview_target }}>
+ {% if page.is_default_language %}Overview{% else %}概览{% endif %}</a></li>
+ {% endif %}
{% assign elements = elements | push: children %}
{% assign elementsPosStack = elementsPosStack | push: elementsPos %}
{% assign posStack = posStack | push: pos %}
diff --git a/docs/dev/api_concepts.zh.md b/docs/dev/api_concepts.zh.md
index bd7ca5a..619ee70 100644
--- a/docs/dev/api_concepts.zh.md
+++ b/docs/dev/api_concepts.zh.md
@@ -783,7 +783,7 @@ defined in the `write()`and `readFields()` methods will be used for serializatio
You can use special types, including Scala's `Either`, `Option`, and `Try`.
The Java API has its own custom implementation of `Either`.
-Similarly to Scala's `Either`, it represents a value of one two possible types, *Left* or *Right*.
+Similarly to Scala's `Either`, it represents a value of two possible types, *Left* or *Right*.
`Either` can be useful for error handling or operators that need to output two different types of records.
#### Type Erasure & Type Inference
diff --git a/docs/dev/connectors/kafka.md b/docs/dev/connectors/kafka.md
index e4e4d5d..8f893c4 100644
--- a/docs/dev/connectors/kafka.md
+++ b/docs/dev/connectors/kafka.md
@@ -88,7 +88,7 @@ For most users, the `FlinkKafkaConsumer08` (part of `flink-connector-kafka`) is
<td>>= 1.0.0</td>
<td>
This universal Kafka connector attempts to track the latest version of the Kafka client.
- The version of the client it uses may change between Flink releases. As of this release, it uses the Kafka 2.2.0 client.
+ The version of the client it uses may change between Flink releases. Starting with Flink 1.9 release, it uses the Kafka 2.2.0 client.
Modern Kafka clients are backwards compatible with broker versions 0.10.0 or later.
However for Kafka 0.11.x and 0.10.x versions, we recommend using dedicated
flink-connector-kafka-0.11{{ site.scala_version_suffix }} and flink-connector-kafka-0.10{{ site.scala_version_suffix }} respectively.
diff --git a/docs/dev/connectors/kafka.zh.md b/docs/dev/connectors/kafka.zh.md
index 2412277..28f3a25 100644
--- a/docs/dev/connectors/kafka.zh.md
+++ b/docs/dev/connectors/kafka.zh.md
@@ -88,7 +88,7 @@ For most users, the `FlinkKafkaConsumer08` (part of `flink-connector-kafka`) is
<td>>= 1.0.0</td>
<td>
This universal Kafka connector attempts to track the latest version of the Kafka client.
- The version of the client it uses may change between Flink releases.
+ The version of the client it uses may change between Flink releases. Starting with Flink 1.9 release, it uses the Kafka 2.2.0 client.
Modern Kafka clients are backwards compatible with broker versions 0.10.0 or later.
However for Kafka 0.11.x and 0.10.x versions, we recommend using dedicated
flink-connector-kafka-0.11{{ site.scala_version_suffix }} and flink-connector-kafka-0.10{{ site.scala_version_suffix }} respectively.
diff --git a/docs/dev/execution.zh.md b/docs/dev/execution.zh.md
index 4f613e0..100d901 100644
--- a/docs/dev/execution.zh.md
+++ b/docs/dev/execution.zh.md
@@ -1,5 +1,5 @@
---
-title: "Managing Execution"
+title: "管理执行"
nav-id: execution
nav-parent_id: dev
nav-pos: 60
diff --git a/docs/dev/execution_configuration.zh.md b/docs/dev/execution_configuration.zh.md
index fc2364b..e32011e 100644
--- a/docs/dev/execution_configuration.zh.md
+++ b/docs/dev/execution_configuration.zh.md
@@ -1,5 +1,5 @@
---
-title: "Execution Configuration"
+title: "执行配置"
nav-parent_id: execution
nav-pos: 10
---
diff --git a/docs/dev/execution_plans.zh.md b/docs/dev/execution_plans.zh.md
index 29a3641..eefa8f4 100644
--- a/docs/dev/execution_plans.zh.md
+++ b/docs/dev/execution_plans.zh.md
@@ -1,5 +1,5 @@
---
-title: "Execution Plans"
+title: "执行计划"
nav-parent_id: execution
nav-pos: 40
---
diff --git a/docs/dev/packaging.zh.md b/docs/dev/packaging.zh.md
index f43ff87..6a6f987 100644
--- a/docs/dev/packaging.zh.md
+++ b/docs/dev/packaging.zh.md
@@ -1,6 +1,6 @@
---
-title: "Program Packaging and Distributed Execution"
-nav-title: Program Packaging
+title: "程序打包和分布式运行"
+nav-title: 程序打包
nav-parent_id: execution
nav-pos: 20
---
diff --git a/docs/dev/parallel.zh.md b/docs/dev/parallel.zh.md
index ae6f863..9031256 100644
--- a/docs/dev/parallel.zh.md
+++ b/docs/dev/parallel.zh.md
@@ -1,5 +1,5 @@
---
-title: "Parallel Execution"
+title: "并发执行"
nav-parent_id: execution
nav-pos: 30
---
diff --git a/docs/dev/restart_strategies.zh.md b/docs/dev/restart_strategies.zh.md
index 5be430e..4b56187 100644
--- a/docs/dev/restart_strategies.zh.md
+++ b/docs/dev/restart_strategies.zh.md
@@ -1,5 +1,5 @@
---
-title: "Restart Strategies"
+title: "重启策略"
nav-parent_id: execution
nav-pos: 50
---
diff --git a/docs/dev/stream/state/queryable_state.zh.md b/docs/dev/stream/state/queryable_state.zh.md
index 5115f11..c9a16c4 100644
--- a/docs/dev/stream/state/queryable_state.zh.md
+++ b/docs/dev/stream/state/queryable_state.zh.md
@@ -33,18 +33,18 @@ under the License.
</div>
In a nutshell, this feature exposes Flink's managed keyed (partitioned) state
-(see [Working with State]({{ site.baseurl }}/dev/stream/state/state.html)) to the outside world and
-allows the user to query a job's state from outside Flink. For some scenarios, queryable state
-eliminates the need for distributed operations/transactions with external systems such as key-value
-stores which are often the bottleneck in practice. In addition, this feature may be particularly
+(see [Working with State]({{ site.baseurl }}/dev/stream/state/state.html)) to the outside world and
+allows the user to query a job's state from outside Flink. For some scenarios, queryable state
+eliminates the need for distributed operations/transactions with external systems such as key-value
+stores which are often the bottleneck in practice. In addition, this feature may be particularly
useful for debugging purposes.
<div class="alert alert-warning">
- <strong>Attention:</strong> When querying a state object, that object is accessed from a concurrent
+ <strong>Attention:</strong> When querying a state object, that object is accessed from a concurrent
thread without any synchronization or copying. This is a design choice, as any of the above would lead
- to increased job latency, which we wanted to avoid. Since any state backend using Java heap space,
- <i>e.g.</i> <code>MemoryStateBackend</code> or <code>FsStateBackend</code>, does not work
- with copies when retrieving values but instead directly references the stored values, read-modify-write
+ to increased job latency, which we wanted to avoid. Since any state backend using Java heap space,
+ <i>e.g.</i> <code>MemoryStateBackend</code> or <code>FsStateBackend</code>, does not work
+ with copies when retrieving values but instead directly references the stored values, read-modify-write
patterns are unsafe and may cause the queryable state server to fail due to concurrent modifications.
The <code>RocksDBStateBackend</code> is safe from these issues.
</div>
@@ -54,25 +54,27 @@ useful for debugging purposes.
Before showing how to use the Queryable State, it is useful to briefly describe the entities that compose it.
The Queryable State feature consists of three main entities:
- 1. the `QueryableStateClient`, which (potentially) runs outside the Flink cluster and submits the user queries,
- 2. the `QueryableStateClientProxy`, which runs on each `TaskManager` (*i.e.* inside the Flink cluster) and is responsible
- for receiving the client's queries, fetching the requested state from the responsible Task Manager on his behalf, and
- returning it to the client, and
+ 1. the `QueryableStateClient`, which (potentially) runs outside the Flink cluster and submits the user queries,
+ 2. the `QueryableStateClientProxy`, which runs on each `TaskManager` (*i.e.* inside the Flink cluster) and is responsible
+ for receiving the client's queries, fetching the requested state from the responsible Task Manager on his behalf, and
+ returning it to the client, and
3. the `QueryableStateServer` which runs on each `TaskManager` and is responsible for serving the locally stored state.
-
-The client connects to one of the proxies and sends a request for the state associated with a specific
-key, `k`. As stated in [Working with State]({{ site.baseurl }}/dev/stream/state/state.html), keyed state is organized in
-*Key Groups*, and each `TaskManager` is assigned a number of these key groups. To discover which `TaskManager` is
-responsible for the key group holding `k`, the proxy will ask the `JobManager`. Based on the answer, the proxy will
+
+The client connects to one of the proxies and sends a request for the state associated with a specific
+key, `k`. As stated in [Working with State]({{ site.baseurl }}/dev/stream/state/state.html), keyed state is organized in
+*Key Groups*, and each `TaskManager` is assigned a number of these key groups. To discover which `TaskManager` is
+responsible for the key group holding `k`, the proxy will ask the `JobManager`. Based on the answer, the proxy will
then query the `QueryableStateServer` running on that `TaskManager` for the state associated with `k`, and forward the
response back to the client.
## Activating Queryable State
-To enable queryable state on your Flink cluster, you just have to copy the
-`flink-queryable-state-runtime{{ site.scala_version_suffix }}-{{site.version }}.jar`
-from the `opt/` folder of your [Flink distribution](https://flink.apache.org/downloads.html "Apache Flink: Downloads"),
-to the `lib/` folder. Otherwise, the queryable state feature is not enabled.
+To enable queryable state on your Flink cluster, you need to do the following:
+
+ 1. copy the `flink-queryable-state-runtime{{ site.scala_version_suffix }}-{{site.version }}.jar`
+from the `opt/` folder of your [Flink distribution](https://flink.apache.org/downloads.html "Apache Flink: Downloads"),
+to the `lib/` folder.
+ 2. set the property `queryable-state.enable` to `true`. See the [Configuration]({{ site.baseurl }}/ops/config.html#queryable-state) documentation for details and additional parameters.
To verify that your cluster is running with queryable state enabled, check the logs of any
task manager for the line: `"Started the Queryable State Proxy Server @ ..."`.
diff --git a/docs/dev/table/functions.zh.md b/docs/dev/table/functions.zh.md
index 2ec5f7c..aaba9ca 100644
--- a/docs/dev/table/functions.zh.md
+++ b/docs/dev/table/functions.zh.md
@@ -2902,7 +2902,7 @@ concat_ws(STRING1, STRING2, STRING3, ...)
{% endhighlight %}
</td>
<td>
- <p>Returns a string that concatenates <i>STRING2, STRING3, ...</i> with a separator <i>STRING1</i>. The separator is added between the strings to be concatenated. Returns NULL If <i>STRING1</i> is NULL. Compared with <code>concat()</code>, <code>concat_ws()</code> automatically skips NULL arguments.</p>
+ <p>Returns a string that concatenates <i>STRING2, STRING3, ...</i> with a separator <i>STRING1</i>. The separator is added between the strings to be concatenated. Returns NULL If <i>STRING1</i> is NULL. Compared with <code>concat()</code>, <code>concat_ws()</code> automatically skips NULL arguments.</p>
<p>E.g., <code>concat_ws('~', 'AA', Null(STRING), 'BB', '', 'CC')</code> returns "AA~BB~~CC".</p>
</td>
</tr>
@@ -2914,7 +2914,7 @@ STRING1.lpad(INT, STRING2)
{% endhighlight %}
</td>
<td>
- <p>Returns a new string from <i>STRING1</i> left-padded with <i>STRING2</i> to a length of <i>INT</i> characters. If the length of <i>STRING1</i> is shorter than <i>INT</i>, returns <i>STRING1</i> shortened to <i>INT</i> characters.</p>
+ <p>Returns a new string from <i>STRING1</i> left-padded with <i>STRING2</i> to a length of <i>INT</i> characters. If the length of <i>STRING1</i> is shorter than <i>INT</i>, returns <i>STRING1</i> shortened to <i>INT</i> characters.</p>
<p>E.g., <code>'hi'.lpad(4, '??')</code> returns "??hi"; <code>'hi'.lpad(1, '??')</code> returns "h".</p>
</td>
</tr>
@@ -2926,11 +2926,11 @@ STRING1.rpad(INT, STRING2)
{% endhighlight %}
</td>
<td>
- <p>Returns a new string from <i>STRING1</i> right-padded with <i>STRING2</i> to a length of <i>INT</i> characters. If the length of <i>STRING1</i> is shorter than <i>INT</i>, returns <i>STRING1</i> shortened to <i>INT</i> characters.</p>
+ <p>Returns a new string from <i>STRING1</i> right-padded with <i>STRING2</i> to a length of <i>INT</i> characters. If the length of <i>STRING1</i> is shorter than <i>INT</i>, returns <i>STRING1</i> shortened to <i>INT</i> characters.</p>
<p>E.g., <code>'hi'.rpad(4, '??')</code> returns "hi??"; <code>'hi'.rpad(1, '??')</code> returns "h".</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight java %}
@@ -2938,11 +2938,11 @@ STRING.fromBase64()
{% endhighlight %}
</td>
<td>
- <p>Returns the base64-decoded result from <i>STRING</i>; returns NULL if <i>STRING</i> is NULL.</p>
+ <p>Returns the base64-decoded result from <i>STRING</i>; returns NULL if <i>STRING</i> is NULL.</p>
<p>E.g., <code>'aGVsbG8gd29ybGQ='.fromBase64()</code> returns "hello world".</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight java %}
@@ -2989,7 +2989,7 @@ STRING.charLength()
<td>
<p>Returns the number of characters in <i>STRING</i>.</p>
</td>
- </tr>
+ </tr>
<tr>
<td>
@@ -3045,7 +3045,7 @@ STRING.ltrim()
{% endhighlight %}
</td>
<td>
- <p>Returns a string that removes the left whitespaces from <i>STRING</i>.</p>
+ <p>Returns a string that removes the left whitespaces from <i>STRING</i>.</p>
<p>E.g., <code>" This is a test String.".ltrim()</code> returns "This is a test String.".</p>
</td>
</tr>
@@ -3057,7 +3057,7 @@ STRING.rtrim()
{% endhighlight %}
</td>
<td>
- <p>Returns a string that removes the right whitespaces from <i>STRING</i>.</p>
+ <p>Returns a string that removes the right whitespaces from <i>STRING</i>.</p>
<p>E.g., <code>"This is a test String. ".rtrim()</code> returns "This is a test String.".</p>
</td>
</tr>
@@ -3069,10 +3069,10 @@ STRING.repeat(INT)
{% endhighlight %}
</td>
<td>
- <p>Returns a string that repeats the base <i>STRING</i> <i>INT</i> times.</p>
+ <p>Returns a string that repeats the base <i>STRING</i> <i>INT</i> times.</p>
<p>E.g., <code>"This is a test String.".repeat(2)</code> returns "This is a test String.This is a test String.".</p>
</td>
- </tr>
+ </tr>
<tr>
<td>
@@ -3081,7 +3081,7 @@ STRING1.regexpReplace(STRING2, STRING3)
{% endhighlight %}
</td>
<td>
- <p>Returns a string from <i>STRING1</i> with all the substrings that match a regular expression <i>STRING2</i> consecutively being replaced with <i>STRING3</i>.</p>
+ <p>Returns a string from <i>STRING1</i> with all the substrings that match a regular expression <i>STRING2</i> consecutively being replaced with <i>STRING3</i>.</p>
<p>E.g. <code>"foobar".regexpReplace("oo|ar", "")</code> returns "fb".</p>
</td>
</tr>
@@ -3146,7 +3146,7 @@ STRING.initCap()
<p>Returns a new form of <i>STRING</i> with the first character of each word converted to uppercase and the rest characters to lowercase. Here a word means a sequences of alphanumeric characters.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -3166,11 +3166,11 @@ concat_ws(STRING1, STRING2, STRING3, ...)
{% endhighlight %}
</td>
<td>
- <p>Returns a string that concatenates <i>STRING2, STRING3, ...</i> with a separator <i>STRING1</i>. The separator is added between the strings to be concatenated. Returns NULL If <i>STRING1</i> is NULL. Compared with <code>concat()</code>, <code>concat_ws()</code> automatically skips NULL arguments.</p>
+ <p>Returns a string that concatenates <i>STRING2, STRING3, ...</i> with a separator <i>STRING1</i>. The separator is added between the strings to be concatenated. Returns NULL If <i>STRING1</i> is NULL. Compared with <code>concat()</code>, <code>concat_ws()</code> automatically skips NULL arguments.</p>
<p>E.g., <code>concat_ws("~", "AA", Null(Types.STRING), "BB", "", "CC")</code> returns "AA~BB~~CC".</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -3178,7 +3178,7 @@ STRING1.lpad(INT, STRING2)
{% endhighlight %}
</td>
<td>
- <p>Returns a new string from <i>STRING1</i> left-padded with <i>STRING2</i> to a length of <i>INT</i> characters. If the length of <i>STRING1</i> is shorter than <i>INT</i>, returns <i>STRING1</i> shortened to <i>INT</i> characters.</p>
+ <p>Returns a new string from <i>STRING1</i> left-padded with <i>STRING2</i> to a length of <i>INT</i> characters. If the length of <i>STRING1</i> is shorter than <i>INT</i>, returns <i>STRING1</i> shortened to <i>INT</i> characters.</p>
<p>E.g., <code>"hi".lpad(4, "??")</code> returns "??hi"; <code>"hi".lpad(1, "??")</code> returns "h".</p>
</td>
</tr>
@@ -3190,11 +3190,11 @@ STRING1.rpad(INT, STRING2)
{% endhighlight %}
</td>
<td>
- <p>Returns a new string from <i>STRING1</i> right-padded with <i>STRING2</i> to a length of <i>INT</i> characters. If the length of <i>STRING1</i> is shorter than <i>INT</i>, returns <i>STRING1</i> shortened to <i>INT</i> characters.</p>
+ <p>Returns a new string from <i>STRING1</i> right-padded with <i>STRING2</i> to a length of <i>INT</i> characters. If the length of <i>STRING1</i> is shorter than <i>INT</i>, returns <i>STRING1</i> shortened to <i>INT</i> characters.</p>
<p>E.g., <code>"hi".rpad(4, "??")</code> returns "hi??"; <code>"hi".rpad(1, "??")</code> returns "h".</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -3202,11 +3202,11 @@ STRING.fromBase64()
{% endhighlight %}
</td>
<td>
- <p>Returns the base64-decoded result from <i>STRING</i>; returns null If <i>STRING</i> is NULL.</p>
+ <p>Returns the base64-decoded result from <i>STRING</i>; returns null If <i>STRING</i> is NULL.</p>
<p>E.g., <code>"aGVsbG8gd29ybGQ=".fromBase64()</code> returns "hello world".</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -3278,7 +3278,7 @@ INTERVAL string range
{% endhighlight %}
</td>
<td>
- <p>Parses an interval <i>string</i> in the form "dd hh:mm:ss.fff" for SQL intervals of milliseconds or "yyyy-mm" for SQL intervals of months. An interval range might be <code>DAY</code>, <code>MINUTE</code>, <code>DAY TO HOUR</code>, or <code>DAY TO SECOND</code> for intervals of milliseconds; <code>YEAR</code> or <code>YEAR TO MONTH</code> for intervals of months.</p>
+ <p>Parses an interval <i>string</i> in the form "dd hh:mm:ss.fff" for SQL intervals of milliseconds or "yyyy-mm" for SQL intervals of months. An interval range might be <code>DAY</code>, <code>MINUTE</code>, <code>DAY TO HOUR</code>, or <code>DAY TO SECOND</code> for intervals of milliseconds; <code>YEAR</code> or <code>YEAR TO MONTH</code> for intervals of months.</p>
<p>E.g., <code>INTERVAL '10 00:00:00.004' DAY TO SECOND</code>, <code>INTERVAL '10' DAY</code>, or <code>INTERVAL '2-10' YEAR TO MONTH</code> return intervals.</p>
</td>
</tr>
@@ -3357,11 +3357,11 @@ YEAR(date)
{% endhighlight %}
</td>
<td>
- <p>Returns the year from SQL date <i>date</i>. Equivalent to EXTRACT(YEAR FROM date).</p>
+ <p>Returns the year from SQL date <i>date</i>. Equivalent to EXTRACT(YEAR FROM date).</p>
<p>E.g., <code>YEAR(DATE '1994-09-27')</code> returns 1994.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight text %}
@@ -3369,7 +3369,7 @@ QUARTER(date)
{% endhighlight %}
</td>
<td>
- <p>Returns the quarter of a year (an integer between 1 and 4) from SQL date <i>date</i>. Equivalent to <code>EXTRACT(QUARTER FROM date)</code>.</p>
+ <p>Returns the quarter of a year (an integer between 1 and 4) from SQL date <i>date</i>. Equivalent to <code>EXTRACT(QUARTER FROM date)</code>.</p>
<p>E.g., <code>QUARTER(DATE '1994-09-27')</code> returns 3.</p>
</td>
</tr>
@@ -3381,7 +3381,7 @@ MONTH(date)
{% endhighlight %}
</td>
<td>
- <p>Returns the month of a year (an integer between 1 and 12) from SQL date <i>date</i>. Equivalent to <code>EXTRACT(MONTH FROM date)</code>.</p>
+ <p>Returns the month of a year (an integer between 1 and 12) from SQL date <i>date</i>. Equivalent to <code>EXTRACT(MONTH FROM date)</code>.</p>
<p>E.g., <code>MONTH(DATE '1994-09-27')</code> returns 9.</p>
</td>
</tr>
@@ -3477,7 +3477,7 @@ FLOOR(timepoint TO timeintervalunit)
{% endhighlight %}
</td>
<td>
- <p>Returns a value that rounds <i>timepoint</i> down to the time unit <i>timeintervalunit</i>.</p>
+ <p>Returns a value that rounds <i>timepoint</i> down to the time unit <i>timeintervalunit</i>.</p>
<p>E.g., <code>FLOOR(TIME '12:44:31' TO MINUTE)</code> returns 12:44:00.</p>
</td>
</tr>
@@ -3524,7 +3524,7 @@ TIMESTAMPADD(timeintervalunit, interval, timepoint)
{% endhighlight %}
</td>
<td>
- <p>Returns a new time value that adds a (signed) integer interval to <i>timepoint</i>. The unit for <i>interval</i> is given by the unit argument, which should be one of the following values: <code>SECOND</code>, <code>MINUTE</code>, <code>HOUR</code>, <code>DAY</code>, <code>WEEK</code>, <code>MONTH</code>, <code>QUARTER</code>, or <code>YEAR</code>.</p>
+ <p>Returns a new time value that adds a (signed) integer interval to <i>timepoint</i>. The unit for <i>interval</i> is given by the unit argument, which should be one of the following values: <code>SECOND</code>, <code>MINUTE</code>, <code>HOUR</code>, <code>DAY</code>, <code>WEEK</code>, <code>MONTH</code>, <code>QUARTER</code>, or <code>YEAR</code>.</p>
<p>E.g., <code>TIMESTAMPADD(WEEK, 1, DATE '2003-01-02')</code> returns <code>2003-01-09</code>.</p>
</td>
</tr>
@@ -3553,7 +3553,7 @@ TIMESTAMPDIFF(timepointunit, timepoint1, timepoint2)
<th class="text-center">Description</th>
</tr>
</thead>
-
+
<tbody>
<tr>
<td>
@@ -3635,7 +3635,7 @@ NUMERIC.weeks
<p>E.g., <code>2.weeks</code> returns 1209600000.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight java %}
@@ -3770,7 +3770,7 @@ TIMEPOINT.floor(TIMEINTERVALUNIT)
{% endhighlight %}
</td>
<td>
- <p>Returns a value that rounds <i>TIMEPOINT</i> down to the time unit <i>TIMEINTERVALUNIT</i>.</p>
+ <p>Returns a value that rounds <i>TIMEPOINT</i> down to the time unit <i>TIMEINTERVALUNIT</i>.</p>
<p>E.g., <code>'12:44:31'.toDate.floor(MINUTE)</code> returns 12:44:00.</p>
</td>
</tr>
@@ -3918,7 +3918,7 @@ NUMERIC.weeks
<p>E.g., <code>2.weeks</code> returns 1209600000.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -4053,7 +4053,7 @@ TIMEPOINT.floor(TIMEINTERVALUNIT)
{% endhighlight %}
</td>
<td>
- <p>Returns a value that rounds <i>TIMEPOINT</i> down to the time unit <i>TIMEINTERVALUNIT</i>.</p>
+ <p>Returns a value that rounds <i>TIMEPOINT</i> down to the time unit <i>TIMEINTERVALUNIT</i>.</p>
<p>E.g., <code>"12:44:31".toDate.floor(TimeIntervalUnit.MINUTE)</code> returns 12:44:00.</p>
</td>
</tr>
@@ -4153,7 +4153,7 @@ END
{% endhighlight %}
</td>
<td>
- <p>Returns <i>resultX</i> when the first <i>conditionX</i> is met.
+ <p>Returns <i>resultX</i> when the first <i>conditionX</i> is met.
When no condition is met, returns <i>resultZ</i> if it is provided and returns NULL otherwise.</p>
</td>
</tr>
@@ -4193,7 +4193,7 @@ COALESCE(value1, value2 [, value3 ]* )
<th class="text-center">Description</th>
</tr>
</thead>
-
+
<tbody>
<tr>
<td>
@@ -4202,7 +4202,7 @@ BOOLEAN.?(VALUE1, VALUE2)
{% endhighlight %}
</td>
<td>
- <p>Returns <i>VALUE1</i> if <i>BOOLEAN</i> evaluates to TRUE; returns <i>VALUE2</i> otherwise.</p>
+ <p>Returns <i>VALUE1</i> if <i>BOOLEAN</i> evaluates to TRUE; returns <i>VALUE2</i> otherwise.</p>
<p>E.g., <code>(42 > 5).?('A', 'B')</code> returns "A".</p>
</td>
</tr>
@@ -4218,7 +4218,7 @@ BOOLEAN.?(VALUE1, VALUE2)
<th class="text-center">Description</th>
</tr>
</thead>
-
+
<tbody>
<tr>
<td>
@@ -4227,7 +4227,7 @@ BOOLEAN.?(VALUE1, VALUE2)
{% endhighlight %}
</td>
<td>
- <p>Returns <i>VALUE1</i> if <i>BOOLEAN</i> evaluates to TRUE; returns <i>VALUE2</i> otherwise.</p>
+ <p>Returns <i>VALUE1</i> if <i>BOOLEAN</i> evaluates to TRUE; returns <i>VALUE2</i> otherwise.</p>
<p>E.g., <code>(42 > 5).?("A", "B")</code> returns "A".</p>
</td>
</tr>
@@ -4275,7 +4275,7 @@ CAST(value AS type)
<th class="text-center">Description</th>
</tr>
</thead>
-
+
<tbody>
<tr>
<td>
@@ -4435,7 +4435,7 @@ ARRAY.element()
<p>Returns the sole element of <i>ARRAY</i> (whose cardinality should be one); returns NULL if <i>ARRAY</i> is empty. Throws an exception if <i>ARRAY</i> has more than one element.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight java %}
@@ -4503,7 +4503,7 @@ ARRAY.element()
<p>Returns the sole element of <i>ARRAY</i> (whose cardinality should be one); returns NULL if <i>ARRAY</i> is empty. Throws an exception if <i>ARRAY</i> has more than one element.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -4589,7 +4589,7 @@ MAP ‘[’ value1, value2 [, value3, value4 ]* ‘]’
<th class="text-center">Description</th>
</tr>
</thead>
-
+
<tbody>
<tr>
<td>
@@ -4601,7 +4601,7 @@ row(ANY1, ANY2, ...)
<p>Returns a row created from a list of object values (<i>ANY1, ANY2</i>, ...). Row is composite type that can be access via <a href="#value-access-functions">value access functions</a>.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight java %}
@@ -4623,7 +4623,7 @@ map(ANY1, ANY2, ANY3, ANY4, ...)
<p>Returns a map created from a list of key-value pairs ((<i>ANY1, ANY2</i>), <i>(ANY3, ANY4)</i>, ...).</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight java %}
@@ -4658,7 +4658,7 @@ row(ANY1, ANY2, ...)
<p>Returns a row created from a list of object values (<i>ANY1, ANY2</i>, ...). Row is composite type that can be access via <a href="#value-access-functions">value access functions</a>.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -4669,7 +4669,7 @@ array(ANY1, ANY2, ...)
<p>Returns an array created from a list of object values (<i>ANY1, ANY2</i>, ...).</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -4680,7 +4680,7 @@ map(ANY1, ANY2, ANY3, ANY4, ...)
<p>Returns a map created from a list of key-value pairs ((<i>ANY1, ANY2</i>), <i>(ANY3, ANY4)</i>, ...).</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -4917,7 +4917,7 @@ SHA1(string)
<p>Returns the SHA-1 hash of <i>string</i> as a string of 40 hexadecimal digits; returns NULL if <i>string</i> is NULL.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight text %}
@@ -4927,8 +4927,8 @@ SHA224(string)
<td>
<p>Returns the SHA-224 hash of <i>string</i> as a string of 56 hexadecimal digits; returns NULL if <i>string</i> is NULL.</p>
</td>
- </tr>
-
+ </tr>
+
<tr>
<td>
{% highlight text %}
@@ -4939,7 +4939,7 @@ SHA256(string)
<p>Returns the SHA-256 hash of <i>string</i> as a string of 64 hexadecimal digits; returns NULL if <i>string</i> is NULL.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight text %}
@@ -4949,7 +4949,7 @@ SHA384(string)
<td>
<p>Returns the SHA-384 hash of <i>string</i> as a string of 96 hexadecimal digits; returns NULL if <i>string</i> is NULL.</p>
</td>
- </tr>
+ </tr>
<tr>
<td>
@@ -4985,7 +4985,7 @@ SHA2(string, hashLength)
<th class="text-center">Description</th>
</tr>
</thead>
-
+
<tbody>
<tr>
<td>
@@ -5076,7 +5076,7 @@ STRING.sha2(INT)
<th class="text-center">Description</th>
</tr>
</thead>
-
+
<tbody>
<tr>
<td>
@@ -5121,7 +5121,7 @@ STRING.sha256()
<p>Returns the SHA-256 hash of <i>STRING</i> as a string of 64 hexadecimal digits; returns NULL if <i>STRING</i> is NULL.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -5131,7 +5131,7 @@ STRING.sha384()
<td>
<p>Returns the SHA-384 hash of <i>STRING</i> as a string of 96 hexadecimal digits; returns NULL if <i>STRING</i> is NULL.</p>
</td>
- </tr>
+ </tr>
<tr>
<td>
@@ -5234,7 +5234,7 @@ ANY.as(NAME1, NAME2, ...)
Aggregate Functions
-------------------
-The aggregate functions take an expression across all the rows as the input and return a single aggregated value as the result.
+The aggregate functions take an expression across all the rows as the input and return a single aggregated value as the result.
<div class="codetabs" markdown="1">
<div data-lang="SQL" markdown="1">
@@ -5380,7 +5380,7 @@ COLLECT([ ALL | DISTINCT ] expression)
<th class="text-center">Description</th>
</tr>
</thead>
-
+
<tbody>
<tr>
<td>
@@ -5458,7 +5458,7 @@ FIELD.stddevPop
<p>Returns the population standard deviation of numeric field <i>FIELD</i> across all input rows.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight java %}
@@ -5592,7 +5592,7 @@ FIELD.stddevPop
<p>Returns the population standard deviation of numeric field <i>FIELD</i> across all input rows.</p>
</td>
</tr>
-
+
<tr>
<td>
{% highlight scala %}
@@ -5763,7 +5763,7 @@ The following table lists specifiers for date format functions.
Time Interval and Point Unit Specifiers
---------------------------------------
-The following table lists specifiers for time interval and time point units.
+The following table lists specifiers for time interval and time point units.
For Table API, please use `_` for spaces (e.g., `DAY_TO_HOUR`).
@@ -5800,3 +5800,285 @@ For Table API, please use `_` for spaces (e.g., `DAY_TO_HOUR`).
| | `SQL_TSI_SECOND ` _(SQL-only)_ |
{% top %}
+
+Column Functions
+---------------------------------------
+
+The column functions are used to select or deselect table columns.
+
+| SYNTAX | DESC |
+| :--------------------- | :-------------------------- |
+| withColumns(...) | select the specified columns |
+| withoutColumns(...) | deselect the columns specified |
+
+The detailed syntax is as follows:
+
+{% highlight text %}
+columnFunction:
+ withColumns(columnExprs)
+ withoutColumns(columnExprs)
+
+columnExprs:
+ columnExpr [, columnExpr]*
+
+columnExpr:
+ columnRef | columnIndex to columnIndex | columnName to columnName
+
+columnRef:
+ columnName(The field name that exists in the table) | columnIndex(a positive integer starting from 1)
+{% endhighlight %}
+
+The usage of the column function is illustrated in the following table. (Suppose we have a table with 5 columns: `(a: Int, b: Long, c: String, d:String, e: String)`):
+
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Api</th>
+ <th class="text-center">Usage</th>
+ <th class="text-center">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>
+ withColumns(*)|*
+ </td>
+ <td>
+{% highlight java %}
+select("withColumns(*)") | select("*") = select("a, b, c, d, e")
+{% endhighlight %}
+ </td>
+ <td>
+ all the columns
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m to n)
+ </td>
+ <td>
+{% highlight java %}
+select("withColumns(2 to 4)") = select("b, c, d")
+{% endhighlight %}
+ </td>
+ <td>
+ columns from m to n
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m, n, k)
+ </td>
+ <td>
+{% highlight java %}
+select("withColumns(1, 3, e)") = select("a, c, e")
+{% endhighlight %}
+ </td>
+ <td>
+ columns m, n, k
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m, n to k)
+ </td>
+ <td>
+{% highlight java %}
+select("withColumns(1, 3 to 5)") = select("a, c, d ,e")
+{% endhighlight %}
+ </td>
+ <td>
+ mixing of the above two representation
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m to n)
+ </td>
+ <td>
+{% highlight java %}
+select("withoutColumns(2 to 4)") = select("a, e")
+{% endhighlight %}
+ </td>
+ <td>
+ deselect columns from m to n
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m, n, k)
+ </td>
+ <td>
+{% highlight java %}
+select("withoutColumns(1, 3, 5)") = select("b, d")
+{% endhighlight %}
+ </td>
+ <td>
+ deselect columns m, n, k
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m, n to k)
+ </td>
+ <td>
+{% highlight java %}
+select("withoutColumns(1, 3 to 5)") = select("b")
+{% endhighlight %}
+ </td>
+ <td>
+ mixing of the above two representation
+ </td>
+ </tr>
+
+ </tbody>
+</table>
+</div>
+
+<div data-lang="scala" markdown="1">
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Api</th>
+ <th class="text-center">Usage</th>
+ <th class="text-center">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>
+ withColumns(*)|*
+ </td>
+ <td>
+{% highlight scala %}
+select(withColumns('*)) | select('*) = select('a, 'b, 'c, 'd, 'e)
+{% endhighlight %}
+ </td>
+ <td>
+ all the columns
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m to n)
+ </td>
+ <td>
+{% highlight scala %}
+select(withColumns(2 to 4)) = select('b, 'c, 'd)
+{% endhighlight %}
+ </td>
+ <td>
+ columns from m to n
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m, n, k)
+ </td>
+ <td>
+{% highlight scala %}
+select(withColumns(1, 3, 'e)) = select('a, 'c, 'e)
+{% endhighlight %}
+ </td>
+ <td>
+ columns m, n, k
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m, n to k)
+ </td>
+ <td>
+{% highlight scala %}
+select(withColumns(1, 3 to 5)) = select('a, 'c, 'd, 'e)
+{% endhighlight %}
+ </td>
+ <td>
+ mixing of the above two representation
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m to n)
+ </td>
+ <td>
+{% highlight scala %}
+select(withoutColumns(2 to 4)) = select('a, 'e)
+{% endhighlight %}
+ </td>
+ <td>
+ deselect columns from m to n
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m, n, k)
+ </td>
+ <td>
+{% highlight scala %}
+select(withoutColumns(1, 3, 5)) = select('b, 'd)
+{% endhighlight %}
+ </td>
+ <td>
+ deselect columns m, n, k
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m, n to k)
+ </td>
+ <td>
+{% highlight scala %}
+select(withoutColumns(1, 3 to 5)) = select('b)
+{% endhighlight %}
+ </td>
+ <td>
+ mixing of the above two representation
+ </td>
+ </tr>
+
+ </tbody>
+</table>
+</div>
+</div>
+
+The column functions can be used in all places where column fields are expected, such as `select, groupBy, orderBy, UDFs etc.` e.g.:
+
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+table
+ .groupBy("withColumns(1 to 3)")
+ .select("withColumns(a to b), myUDAgg(myUDF(withColumns(5 to 20)))")
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+table
+ .groupBy(withColumns(1 to 3))
+ .select(withColumns('a to 'b), myUDAgg(myUDF(withColumns(5 to 20))))
+{% endhighlight %}
+</div>
+</div>
+
+<span class="label label-info">Note</span> Column functions are only used in Table API.
+
+{% top %}
diff --git a/docs/dev/table/tableApi.zh.md b/docs/dev/table/tableApi.zh.md
index 4f75740..4aa1c98 100644
--- a/docs/dev/table/tableApi.zh.md
+++ b/docs/dev/table/tableApi.zh.md
@@ -1895,6 +1895,7 @@ The `OverWindow` defines a range of rows over which aggregates are computed. `Ov
### Row-based Operations
+The row-based operations generate outputs with multiple columns.
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
<table class="table table-bordered">
@@ -1913,15 +1914,168 @@ The `OverWindow` defines a range of rows over which aggregates are computed. `Ov
<td>
<p>Performs a map operation with a user-defined scalar function or built-in scalar function. The output will be flattened if the output type is a composite type.</p>
{% highlight java %}
+public class MyMapFunction extends ScalarFunction {
+ public Row eval(String a) {
+ return Row.of(a, "pre-" + a);
+ }
+
+ @Override
+ public TypeInformation<?> getResultType(Class<?>[] signature) {
+ return Types.ROW(Types.STRING(), Types.STRING());
+ }
+}
+
ScalarFunction func = new MyMapFunction();
tableEnv.registerFunction("func", func);
Table table = input
- .map(func("c")).as("a, b")
+ .map("func(c)").as("a, b")
{% endhighlight %}
</td>
</tr>
+ <tr>
+ <td>
+ <strong>FlatMap</strong><br>
+ <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span>
+ </td>
+ <td>
+ <p>Performs a flatMap operation with a table function.</p>
+{% highlight java %}
+public class MyFlatMapFunction extends TableFunction<Row> {
+
+ public void eval(String str) {
+ if (str.contains("#")) {
+ String[] array = str.split("#");
+ for (int i = 0; i < array.length; ++i) {
+ collect(Row.of(array[i], array[i].length()));
+ }
+ }
+ }
+
+ @Override
+ public TypeInformation<Row> getResultType() {
+ return Types.ROW(Types.STRING(), Types.INT());
+ }
+}
+
+TableFunction func = new MyFlatMapFunction();
+tableEnv.registerFunction("func", func);
+
+Table table = input
+ .flatMap("func(c)").as("a, b")
+{% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ <strong>Aggregate</strong><br>
+ <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span>
+ <span class="label label-info">Result Updating</span>
+ </td>
+ <td>
+ <p>Performs an aggregate operation with an aggregate function. You have to close the "aggregate" with a select statement and the select statement does not support aggregate functions. The output of aggregate will be flattened if the output type is a composite type.</p>
+{% highlight java %}
+public class MyMinMaxAcc {
+ public int min = 0;
+ public int max = 0;
+}
+
+public class MyMinMax extends AggregateFunction<Row, MyMinMaxAcc> {
+
+ public void accumulate(MyMinMaxAcc acc, int value) {
+ if (value < acc.min) {
+ acc.min = value;
+ }
+ if (value > acc.max) {
+ acc.max = value;
+ }
+ }
+
+ @Override
+ public MyMinMaxAcc createAccumulator() {
+ return new MyMinMaxAcc();
+ }
+
+ public void resetAccumulator(MyMinMaxAcc acc) {
+ acc.min = 0;
+ acc.max = 0;
+ }
+
+ @Override
+ public Row getValue(MyMinMaxAcc acc) {
+ return Row.of(acc.min, acc.max);
+ }
+
+ @Override
+ public TypeInformation<Row> getResultType() {
+ return new RowTypeInfo(Types.INT, Types.INT);
+ }
+}
+
+AggregateFunction myAggFunc = new MyMinMax();
+tableEnv.registerFunction("myAggFunc", myAggFunc);
+Table table = input
+ .groupBy("key")
+ .aggregate("myAggFunc(a) as (x, y)")
+ .select("key, x, y")
+{% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ <strong>GroupBy TableAggregation</strong><br>
+ <span class="label label-primary">Streaming</span><br>
+ <span class="label label-info">Result Updating</span>
+ </td>
+ <td>
+ <p>Similar to a <b>GroupBy Aggregation</b>. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
+{% highlight java %}
+ public class MyMinMaxAcc {
+ public int min = 0;
+ public int max = 0;
+ }
+
+ public class MyMinMax extends TableAggregateFunction<Row, MyMinMaxAcc> {
+
+ public void accumulate(MyMinMaxAcc acc, int value) {
+ if (value < acc.min) {
+ acc.min = value;
+ }
+ if (value > acc.max) {
+ acc.max = value;
+ }
+ }
+
+ @Override
+ public MyMinMaxAcc createAccumulator() {
+ return new MyMinMaxAcc();
+ }
+
+ public void emitValue(MyMinMaxAcc acc, Collector<Row> out) {
+ out.collect(Row.of(acc.min, acc.min));
+ out.collect(Row.of(acc.max, acc.max));
+ }
+
+ @Override
+ public TypeInformation<Row> getResultType() {
+ return new RowTypeInfo(Types.INT, Types.INT);
+ }
+ }
+
+TableAggregateFunction tableAggFunc = new MyMinMax();
+tableEnv.registerFunction("myTableAggFunc", tableAggFunc);
+Table orders = tableEnv.scan("Orders");
+Table result = orders
+ .groupBy("a")
+ .flatAggregate("myTableAggFunc(a) as (x, y)")
+ .select("a, x, y");
+{% endhighlight %}
+ <p><b>Note:</b> For streaming queries, the required state to compute the query result might grow infinitely depending on the type of aggregation and the number of distinct grouping keys. Please provide a query configuration with a valid retention interval to prevent excessive state size. See <a href="streaming/query_configuration.html">Query Configuration</a> for details.</p>
+ </td>
+ </tr>
</tbody>
</table>
</div>
@@ -1943,14 +2097,150 @@ Table table = input
<td>
<p>Performs a map operation with a user-defined scalar function or built-in scalar function. The output will be flattened if the output type is a composite type.</p>
{% highlight scala %}
-val func: ScalarFunction = new MyMapFunction()
+class MyMapFunction extends ScalarFunction {
+ def eval(a: String): Row = {
+ Row.of(a, "pre-" + a)
+ }
+
+ override def getResultType(signature: Array[Class[_]]): TypeInformation[_] =
+ Types.ROW(Types.STRING, Types.STRING)
+}
+val func = new MyMapFunction()
val table = input
.map(func('c)).as('a, 'b)
{% endhighlight %}
</td>
</tr>
+ <tr>
+ <td>
+ <strong>FlatMap</strong><br>
+ <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span>
+ </td>
+ <td>
+ <p>Performs a flatMap operation with a table function.</p>
+{% highlight scala %}
+class MyFlatMapFunction extends TableFunction[Row] {
+ def eval(str: String): Unit = {
+ if (str.contains("#")) {
+ str.split("#").foreach({ s =>
+ val row = new Row(2)
+ row.setField(0, s)
+ row.setField(1, s.length)
+ collect(row)
+ })
+ }
+ }
+
+ override def getResultType: TypeInformation[Row] = {
+ Types.ROW(Types.STRING, Types.INT)
+ }
+}
+
+val func = new MyFlatMapFunction
+val table = input
+ .flatMap(func('c)).as('a, 'b)
+{% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ <strong>Aggregate</strong><br>
+ <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span>
+ <span class="label label-info">Result Updating</span>
+ </td>
+ <td>
+ <p>Performs an aggregate operation with an aggregate function. You have to close the "aggregate" with a select statement and the select statement does not support aggregate functions. The output of aggregate will be flattened if the output type is a composite type.</p>
+{% highlight scala %}
+case class MyMinMaxAcc(var min: Int, var max: Int)
+
+class MyMinMax extends AggregateFunction[Row, MyMinMaxAcc] {
+
+ def accumulate(acc: MyMinMaxAcc, value: Int): Unit = {
+ if (value < acc.min) {
+ acc.min = value
+ }
+ if (value > acc.max) {
+ acc.max = value
+ }
+ }
+
+ override def createAccumulator(): MyMinMaxAcc = MyMinMaxAcc(0, 0)
+
+ def resetAccumulator(acc: MyMinMaxAcc): Unit = {
+ acc.min = 0
+ acc.max = 0
+ }
+
+ override def getValue(acc: MyMinMaxAcc): Row = {
+ Row.of(Integer.valueOf(acc.min), Integer.valueOf(acc.max))
+ }
+
+ override def getResultType: TypeInformation[Row] = {
+ new RowTypeInfo(Types.INT, Types.INT)
+ }
+}
+
+val myAggFunc: AggregateFunction = new MyMinMax
+val table = input
+ .groupBy('key)
+ .aggregate(myAggFunc('a) as ('x, 'y))
+ .select('key, 'x, 'y)
+{% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ <strong>GroupBy TableAggregation</strong><br>
+ <span class="label label-primary">Streaming</span><br>
+ <span class="label label-info">Result Updating</span>
+ </td>
+ <td>
+ <p>Similar to a <b>GroupBy Aggregation</b>. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
+{% highlight scala %}
+case class MyMinMaxAcc(var min: Int, var max: Int)
+
+class MyMinMax extends TableAggregateFunction[Row, MyMinMaxAcc] {
+
+ def accumulate(acc: MyMinMaxAcc, value: Int): Unit = {
+ if (value < acc.min) {
+ acc.min = value
+ }
+ if (value > acc.max) {
+ acc.max = value
+ }
+ }
+
+ def resetAccumulator(acc: MyMinMaxAcc): Unit = {
+ acc.min = 0
+ acc.max = 0
+ }
+
+ override def createAccumulator(): MyMinMaxAcc = MyMinMaxAcc(0, 0)
+
+ def emitValue(acc: MyMinMaxAcc, out: Collector[Row]): Unit = {
+ out.collect(Row.of(Integer.valueOf(acc.min), Integer.valueOf(acc.min)))
+ out.collect(Row.of(Integer.valueOf(acc.max), Integer.valueOf(acc.max)))
+ }
+
+ override def getResultType: TypeInformation[Row] = {
+ new RowTypeInfo(Types.INT, Types.INT)
+ }
+}
+
+val tableAggFunc = new MyMinMax
+val orders: Table = tableEnv.scan("Orders")
+val result = orders
+ .groupBy('a)
+ .flatAggregate(tableAggFunc('a) as ('x, 'y))
+ .select('a, 'x, 'y)
+{% endhighlight %}
+ <p><b>Note:</b> For streaming queries, the required state to compute the query result might grow infinitely depending on the type of aggregation and the number of distinct grouping keys. Please provide a query configuration with a valid retention interval to prevent excessive state size. See <a href="streaming/query_configuration.html">Query Configuration</a> for details.</p>
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/ops/deployment/yarn_setup.zh.md b/docs/ops/deployment/yarn_setup.zh.md
index 9b02576..94ea422 100644
--- a/docs/ops/deployment/yarn_setup.zh.md
+++ b/docs/ops/deployment/yarn_setup.zh.md
@@ -101,6 +101,7 @@ Usage:
-d,--detached Start detached
-jm,--jobManagerMemory <arg> Memory for JobManager Container with optional unit (default: MB)
-nm,--name Set a custom name for the application on YARN
+ -at,--applicationType Set a custom application type on YARN
-q,--query Display available YARN resources (memory, cores)
-qu,--queue <arg> Specify YARN queue.
-s,--slots <arg> Number of slots per TaskManager
diff --git a/docs/ops/index.zh.md b/docs/ops/index.zh.md
index 8be615e..4d6dbf1 100644
--- a/docs/ops/index.zh.md
+++ b/docs/ops/index.zh.md
@@ -1,7 +1,7 @@
---
title: "部署与运维"
nav-id: ops
-nav-title: '<i class="fa fa-sliders title maindish" aria-hidden="true"></i> Deployment & Operations'
+nav-title: '<i class="fa fa-sliders title maindish" aria-hidden="true"></i> 部署与运维'
nav-parent_id: root
nav-pos: 6
---