[flink] branch release-1.9 updated: [FLINK-17254][python][docs] Improve the PyFlink documentation and examples to use SQL DDL for source/sink definition.

[di...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dianfu pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.9 by this push:
     new 4b45e8b  [FLINK-17254][python][docs] Improve the PyFlink documentation and examples to use SQL DDL for source/sink definition.
4b45e8b is described below

commit 4b45e8b3aa07eb37355cf7752b288fc7ed1bf2d2
Author: acqua.csq <ac...@alibaba-inc.com>
AuthorDate: Wed Apr 29 18:38:48 2020 +0800

    [FLINK-17254][python][docs] Improve the PyFlink documentation and examples to use SQL DDL for source/sink definition.
    
    This closes #11945.
---
 docs/tutorials/python_table_api.md                 | 27 ++++++++++++++++++++++
 .../pyflink/table/examples/batch/word_count.py     | 22 +++++++++---------
 2 files changed, 38 insertions(+), 11 deletions(-)

diff --git a/docs/tutorials/python_table_api.md b/docs/tutorials/python_table_api.md
index 7c3076e..44cf030 100644
--- a/docs/tutorials/python_table_api.md
+++ b/docs/tutorials/python_table_api.md
@@ -81,6 +81,33 @@ t_env.connect(FileSystem().path('/tmp/output')) \
     .register_table_sink('mySink')
 {% endhighlight %}
 
+You can also use the TableEnvironment.sql_update() method to register a source/sink table defined in DDL:
+{% highlight python %}
+my_source_ddl = """
+    create table mySource (
+        word VARCHAR
+    ) with (
+        'connector.type' = 'filesystem',
+        'format.type' = 'csv',
+        'connector.path' = '/tmp/input'
+    )
+"""
+
+my_sink_ddl = """
+    create table mySink (
+        word VARCHAR,
+        `count` BIGINT
+    ) with (
+        'connector.type' = 'filesystem',
+        'format.type' = 'csv',
+        'connector.path' = '/tmp/output'
+    )
+"""
+
+t_env.sql_update(my_source_ddl)
+t_env.sql_update(my_sink_ddl)
+{% endhighlight %}
+
 This registers a table named `mySource` and a table named `mySink` in the
 `ExecutionEnvironment`. The table `mySource` has only one column: word.
 It represents the words read from file `/tmp/input`. The table `mySink` has two columns:
diff --git a/flink-python/pyflink/table/examples/batch/word_count.py b/flink-python/pyflink/table/examples/batch/word_count.py
index 39d9958..720e88f 100644
--- a/flink-python/pyflink/table/examples/batch/word_count.py
+++ b/flink-python/pyflink/table/examples/batch/word_count.py
@@ -23,8 +23,6 @@ import tempfile
 
 from pyflink.dataset import ExecutionEnvironment
 from pyflink.table import BatchTableEnvironment, TableConfig
-from pyflink.table.descriptors import FileSystem, OldCsv, Schema
-from pyflink.table.types import DataTypes
 
 
 def word_count():
@@ -54,15 +52,17 @@ def word_count():
 
     logging.info("Results directory: %s", result_path)
 
-    t_env.connect(FileSystem().path(result_path)) \
-        .with_format(OldCsv()
-                     .field_delimiter(',')
-                     .field("word", DataTypes.STRING())
-                     .field("count", DataTypes.BIGINT())) \
-        .with_schema(Schema()
-                     .field("word", DataTypes.STRING())
-                     .field("count", DataTypes.BIGINT())) \
-        .register_table_sink("Results")
+    sink_ddl = """
+        create table Results(
+            word VARCHAR,
+            `count` BIGINT
+        ) with (
+            'connector.type' = 'filesystem',
+            'format.type' = 'csv',
+            'connector.path' = '{}'
+        )
+        """.format(result_path)
+    t_env.sql_update(sink_ddl)
 
     elements = [(word, 1) for word in content.split(" ")]
     t_env.from_elements(elements, ["word", "count"]) \