apache · bowenli86 · Nov 15, 2018 · Dec 2, 2018 · Dec 4, 2018 · Dec 4, 2018
diff --git a/docs/_includes/generated/checkpointing_configuration.html b/docs/_includes/generated/checkpointing_configuration.html
@@ -30,7 +30,7 @@
         <tr>
             <td><h5>state.backend.local-recovery</h5></td>
             <td style="word-wrap: break-word;">false</td>
-            <td></td>
+            <td>This option configures local recovery for this state backend. By default, local recovery is deactivated. Local recovery currently only covers keyed state backends. Currently, MemoryStateBackend does not support local recovery and ignore this option.</td>
         </tr>
         <tr>
             <td><h5>state.checkpoints.dir</h5></td>
@@ -50,7 +50,7 @@
         <tr>
             <td><h5>taskmanager.state.local.root-dirs</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
-            <td></td>
+            <td>The config parameter defining the root directories for storing file-based state for local recovery. Local recovery currently only covers keyed state backends. Currently, MemoryStateBackend does not support local recovery and ignore this option</td>
         </tr>
     </tbody>
 </table>
diff --git a/docs/dev/batch/dataset_transformations.md b/docs/dev/batch/dataset_transformations.md
@@ -712,7 +712,7 @@ class MyCombinableGroupReducer
     out: Collector[String]): Unit =
   {
     val r: (String, Int) =
-      in.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
+      in.iterator.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
     // concat key and sum and emit
     out.collect (r._1 + "-" + r._2)
   }
@@ -722,7 +722,7 @@ class MyCombinableGroupReducer
     out: Collector[(String, Int)]): Unit =
   {
     val r: (String, Int) =
-      in.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
+      in.iterator.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
     // emit tuple with key and sum
     out.collect(r)
   }

diff --git a/docs/dev/batch/index.md b/docs/dev/batch/index.md
@@ -401,12 +401,14 @@ DataSet<Integer> result = in.partitionByRange(0)
     <tr>
       <td><strong>Custom Partitioning</strong></td>
       <td>
-        <p>Manually specify a partitioning over the data.
+        <p>Assigns records based on a key to a specific partition using a custom Partitioner function. 
+          The key can be specified as position key, expression key, and key selector function.
           <br/>
-          <i>Note</i>: This method works only on single field keys.</p>
+          <i>Note</i>: This method only works with a single field key.</p>
 {% highlight java %}
 DataSet<Tuple2<String,Integer>> in = // [...]
-DataSet<Integer> result = in.partitionCustom(Partitioner<K> partitioner, key)
+DataSet<Integer> result = in.partitionCustom(partitioner, key)
+                            .mapPartition(new PartitionMapper());
 {% endhighlight %}
       </td>
     </tr>
@@ -703,13 +705,14 @@ val result = in.partitionByRange(0).mapPartition { ... }
     <tr>
       <td><strong>Custom Partitioning</strong></td>
       <td>
-        <p>Manually specify a partitioning over the data.
+        <p>Assigns records based on a key to a specific partition using a custom Partitioner function. 
+          The key can be specified as position key, expression key, and key selector function.
           <br/>
-          <i>Note</i>: This method works only on single field keys.</p>
+          <i>Note</i>: This method only works with a single field key.</p>
 {% highlight scala %}
 val in: DataSet[(Int, String)] = // [...]
 val result = in
-  .partitionCustom(partitioner: Partitioner[K], key)
+  .partitionCustom(partitioner, key).mapPartition { ... }
 {% endhighlight %}
       </td>
     </tr>

diff --git a/docs/dev/connectors/streamfile_sink.md b/docs/dev/connectors/streamfile_sink.md
@@ -60,17 +60,14 @@ Basic usage thus looks like this:
 <div class="codetabs" markdown="1">
 <div data-lang="java" markdown="1">
 {% highlight java %}
-import org.apache.flink.api.common.serialization.Encoder;
+import org.apache.flink.api.common.serialization.SimpleStringEncoder;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
 
 DataStream<String> input = ...;
 
 final StreamingFileSink<String> sink = StreamingFileSink
-	.forRowFormat(new Path(outputPath), (Encoder<String>) (element, stream) -> {
-		PrintStream out = new PrintStream(stream);
-		out.println(element.f1);
-	})
+	.forRowFormat(new Path(outputPath), new SimpleStringEncoder<>("UTF-8"))
 	.build();
 
 input.addSink(sink);
@@ -79,19 +76,16 @@ input.addSink(sink);
 </div>
 <div data-lang="scala" markdown="1">
 {% highlight scala %}
-import org.apache.flink.api.common.serialization.Encoder
+import org.apache.flink.api.common.serialization.SimpleStringEncoder
 import org.apache.flink.core.fs.Path
 import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
 
 val input: DataStream[String] = ...
 
-final StreamingFileSink[String] sink = StreamingFileSink
-	.forRowFormat(new Path(outputPath), (element, stream) => {
-		val out = new PrintStream(stream)
-		out.println(element.f1)
-	})
-	.build()
-
+val sink: StreamingFileSink[String] = StreamingFileSink
+    .forRowFormat(new Path(outputPath), new SimpleStringEncoder[String]("UTF-8"))
+    .build()
+
 input.addSink(sink)
 
 {% endhighlight %}

diff --git a/docs/dev/libs/ml/quickstart.md b/docs/dev/libs/ml/quickstart.md
@@ -153,6 +153,8 @@ A conversion can be done using a simple normalizer mapping function:
 
 {% highlight scala %}
 
+import org.apache.flink.ml.math.Vector
+
 def normalizer : LabeledVector => LabeledVector = { 
     lv => LabeledVector(if (lv.label > 0.0) 1.0 else -1.0, lv.vector)
 }

diff --git a/docs/dev/stream/operators/index.md b/docs/dev/stream/operators/index.md
@@ -434,14 +434,14 @@ IterativeStream<Long> iteration = initialStream.iterate();
 DataStream<Long> iterationBody = iteration.map (/*do something*/);
 DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){
     @Override
-    public boolean filter(Integer value) throws Exception {
+    public boolean filter(Long value) throws Exception {
         return value > 0;
     }
 });
 iteration.closeWith(feedback);
 DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){
     @Override
-    public boolean filter(Integer value) throws Exception {
+    public boolean filter(Long value) throws Exception {
         return value <= 0;
     }
 });

diff --git a/docs/dev/stream/operators/joining.md b/docs/dev/stream/operators/joining.md
@@ -101,7 +101,7 @@ orangeStream.join(greenStream)
 </div>
 
 ## Sliding Window Join
-When performing a sliding window join, all elements with a common key and common sliding window are joined are pairwise combinations and passed on to the `JoinFunction` or `FlatJoinFunction`. Elements of one stream that do not have elements from the other stream in the current sliding window are not emitted! Note that some elements might be joined in one sliding window but not in another!
+When performing a sliding window join, all elements with a common key and common sliding window are joined as pairwise combinations and passed on to the `JoinFunction` or `FlatJoinFunction`. Elements of one stream that do not have elements from the other stream in the current sliding window are not emitted! Note that some elements might be joined in one sliding window but not in another!
 
 <img src="{{ site.baseurl }}/fig/sliding-window-join.svg" class="center" style="width: 80%;" />
 

diff --git a/docs/dev/stream/operators/windows.md b/docs/dev/stream/operators/windows.md
@@ -783,7 +783,7 @@ When the window is closed, the `ProcessWindowFunction` will be provided with the
 This allows it to incrementally compute windows while having access to the
 additional window meta information of the `ProcessWindowFunction`.
 
-<span class="label label-info">Note</span> You can also the legacy `WindowFunction` instead of
+<span class="label label-info">Note</span> You can also use the legacy `WindowFunction` instead of
 `ProcessWindowFunction` for incremental window aggregation.
 
 #### Incremental Window Aggregation with ReduceFunction
@@ -1034,7 +1034,7 @@ different keys and events for all of them currently fall into the *[12:00, 13:00
 then there will be 1000 window instances that each have their own keyed per-window state.
 
 There are two methods on the `Context` object that a `process()` invocation receives that allow
-access two the two types of state:
+access to the two types of state:
 
  - `globalState()`, which allows access to keyed state that is not scoped to a window
  - `windowState()`, which allows access to keyed state that is also scoped to the window

diff --git a/docs/dev/stream/state/broadcast_state.md b/docs/dev/stream/state/broadcast_state.md
@@ -94,7 +94,7 @@ The exact type of the function depends on the type of the non-broadcasted stream
 </div>
 
 {% highlight java %}
-DataStream<Match> output = colorPartitionedStream
+DataStream<String> output = colorPartitionedStream
                  .connect(ruleBroadcastStream)
                  .process(
 
@@ -107,7 +107,7 @@ DataStream<Match> output = colorPartitionedStream
                      new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
                          // my matching logic
                      }
-                 )
+                 );
 {% endhighlight %}
 
 ### BroadcastProcessFunction and KeyedBroadcastProcessFunction

diff --git a/docs/dev/stream/state/schema_evolution.md b/docs/dev/stream/state/schema_evolution.md
@@ -74,7 +74,7 @@ serialization schema than the previous serializer; if so, the previous serialize
 and written back to bytes again with the new serializer.
 
 Further details about the migration process is out of the scope of this documentation; please refer to
-[here]({{ site.baseurl }}/dev/stream/state/custom_serialization).
+[here]({{ site.baseurl }}/dev/stream/state/custom_serialization.html).
 
 ## Supported data types for schema evolution
 

diff --git a/docs/dev/stream/state/state.md b/docs/dev/stream/state/state.md
@@ -142,7 +142,7 @@ is available in a `RichFunction` has these methods for accessing state:
 * `ValueState<T> getState(ValueStateDescriptor<T>)`
 * `ReducingState<T> getReducingState(ReducingStateDescriptor<T>)`
 * `ListState<T> getListState(ListStateDescriptor<T>)`
-* `AggregatingState<IN, OUT> getAggregatingState(AggregatingState<IN, OUT>)`
+* `AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)`
 * `FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC>)`
 * `MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)`
 

diff --git a/docs/dev/table/connect.md b/docs/dev/table/connect.md
@@ -34,7 +34,7 @@ This page describes how to declare built-in table sources and/or table sinks and
 Dependencies
 ------------
 
-The following table list all available connectors and formats. Their mutual compatibility is tagged in the corresponding sections for [table connectors](connect.html#table-connectors) and [table formats](connect.html#table-formats). The following table provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.
+The following tables list all available connectors and formats. Their mutual compatibility is tagged in the corresponding sections for [table connectors](connect.html#table-connectors) and [table formats](connect.html#table-formats). The following tables provide dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.
 
 {% if site.is_stable %}
 
@@ -43,7 +43,7 @@ The following table list all available connectors and formats. Their mutual comp
 | Name              | Version             | Maven dependency             | SQL Client JAR         |
 | :---------------- | :------------------ | :--------------------------- | :----------------------|
 | Filesystem        |                     | Built-in                     | Built-in               |
-| Elasticsearch     | 6                   | `flink-connector-elasticsearch6` | [Download](http://central.maven.org/maven2/org/apache/flink/flink-connector-elasticsearch6{{site.scala_version_suffix}}/{{site.version}}/flink-connector-elasticsearch6{{site.scala_version_suffix}}-{{site.version}}-sql-jar.jar) |
+| Elasticsearch     | 6                   | `flink-connector-elasticsearch6` | [Download](http://central.maven.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch6{{site.scala_version_suffix}}/{{site.version}}/flink-sql-connector-elasticsearch6{{site.scala_version_suffix}}-{{site.version}}.jar) |
 | Apache Kafka      | 0.8                 | `flink-connector-kafka-0.8`  | Not available          |
 | Apache Kafka      | 0.9                 | `flink-connector-kafka-0.9`  | [Download](http://central.maven.org/maven2/org/apache/flink/flink-connector-kafka-0.9{{site.scala_version_suffix}}/{{site.version}}/flink-connector-kafka-0.9{{site.scala_version_suffix}}-{{site.version}}-sql-jar.jar) |
 | Apache Kafka      | 0.10                | `flink-connector-kafka-0.10` | [Download](http://central.maven.org/maven2/org/apache/flink/flink-connector-kafka-0.10{{site.scala_version_suffix}}/{{site.version}}/flink-connector-kafka-0.10{{site.scala_version_suffix}}-{{site.version}}-sql-jar.jar) |
@@ -60,7 +60,7 @@ The following table list all available connectors and formats. Their mutual comp
 
 {% else %}
 
-This table is only available for stable releases.
+These tables are only available for stable releases.
 
 {% endif %}
 
@@ -145,7 +145,7 @@ tableEnvironment
         "      {\"name\": \"user\", \"type\": \"long\"}," +
         "      {\"name\": \"message\", \"type\": [\"string\", \"null\"]}" +
         "    ]" +
-        "}" +
+        "}"
       )
   )
 
@@ -154,7 +154,7 @@ tableEnvironment
     new Schema()
       .field("rowtime", Types.SQL_TIMESTAMP)
         .rowtime(new Rowtime()
-          .timestampsFromField("ts")
+          .timestampsFromField("timestamp")
           .watermarksPeriodicBounded(60000)
         )
       .field("user", Types.LONG)
@@ -1166,7 +1166,7 @@ ClusterBuilder builder = ... // configure Cassandra cluster connection
 CassandraAppendTableSink sink = new CassandraAppendTableSink(
   builder,
   // the query must match the schema of the table
-  INSERT INTO flink.myTable (id, name, value) VALUES (?, ?, ?));
+  "INSERT INTO flink.myTable (id, name, value) VALUES (?, ?, ?)");
 
 tableEnv.registerTableSink(
   "cassandraOutputTable",
@@ -1187,7 +1187,7 @@ val builder: ClusterBuilder = ... // configure Cassandra cluster connection
 val sink: CassandraAppendTableSink = new CassandraAppendTableSink(
   builder,
   // the query must match the schema of the table
-  INSERT INTO flink.myTable (id, name, value) VALUES (?, ?, ?))
+  "INSERT INTO flink.myTable (id, name, value) VALUES (?, ?, ?)")
 
 tableEnv.registerTableSink(
   "cassandraOutputTable",

diff --git a/docs/dev/table/functions.md b/docs/dev/table/functions.md
@@ -3487,7 +3487,7 @@ DATE_FORMAT(timestamp, string)
 {% endhighlight %}
       </td>
       <td>
-        <p>Returns a string that formats <i>timestamp</i> with a specified format <i>string</i>. The format specification is given in the <a href="#date-format-specifiers">Date Format Specifier table</a>.</p>
+        <p><span class="label label-danger">Attention</span> This function has serious bugs and should not be used for now. Please implement a custom UDF instead or use EXTRACT as a workaround.</p>
       </td>
     </tr>
 
@@ -3756,8 +3756,7 @@ dateFormat(TIMESTAMP, STRING)
 {% endhighlight %}
       </td>
       <td>
-        <p>Returns a string that formats <i>TIMESTAMP</i> with a specified format <i>STRING</i>. The format specification is given in the <a href="#date-format-specifiers">Date Format Specifier table</a>.</p>
-        <p>E.g., <code>dateFormat(ts, '%Y, %d %M')</code> results in strings formatted as "2017, 05 May".</p>
+        <p><span class="label label-danger">Attention</span> This function has serious bugs and should not be used for now. Please implement a custom UDF instead or use extract() as a workaround.</p>
       </td>
     </tr>
 
@@ -4014,8 +4013,7 @@ dateFormat(TIMESTAMP, STRING)
 {% endhighlight %}
       </td>
       <td>
-        <p>Returns a string that formats <i>TIMESTAMP</i> with a specified format <i>STRING</i>. The format specification is given in the <a href="#date-format-specifiers">Date Format Specifier table</a>.</p>
-        <p>E.g., <code>dateFormat('ts, "%Y, %d %M")</code> results in strings formatted as "2017, 05 May".</p>
+        <p><span class="label label-danger">Attention</span> This function has serious bugs and should not be used for now. Please implement a custom UDF instead or use extract() as a workaround.</p>
       </td>
     </tr>
 

diff --git a/docs/dev/table/sql.md b/docs/dev/table/sql.md
@@ -22,7 +22,7 @@ specific language governing permissions and limitations
 under the License.
 -->
 
-SQL queries are specified with the `sqlQuery()` method of the `TableEnvironment`. The method returns the result of the SQL query as a `Table`. A `Table` can be used in [subsequent SQL and Table API queries](common.html#mixing-table-api-and-sql), be [converted into a DataSet or DataStream](common.html#integration-with-datastream-and-dataset-api), or [written to a TableSink](common.html#emit-a-table)). SQL and Table API queries can seamlessly mixed and are holistically optimized and translated into a single program.
+SQL queries are specified with the `sqlQuery()` method of the `TableEnvironment`. The method returns the result of the SQL query as a `Table`. A `Table` can be used in [subsequent SQL and Table API queries](common.html#mixing-table-api-and-sql), be [converted into a DataSet or DataStream](common.html#integration-with-datastream-and-dataset-api), or [written to a TableSink](common.html#emit-a-table)). SQL and Table API queries can be seamlessly mixed and are holistically optimized and translated into a single program.
 
 In order to access a table in a SQL query, it must be [registered in the TableEnvironment](common.html#register-tables-in-the-catalog). A table can be registered from a [TableSource](common.html#register-a-tablesource), [Table](common.html#register-a-table), [DataStream, or DataSet](common.html#register-a-datastream-or-dataset-as-table). Alternatively, users can also [register external catalogs in a TableEnvironment](common.html#register-an-external-catalog) to specify the location of the data sources.