You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2019/06/19 09:41:32 UTC
[flink] branch master updated: [FLINK-12720][python][docs] Add the
Python Table API Sphinx docs
This is an automated email from the ASF dual-hosted git repository.
jincheng 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 d70e8e2 [FLINK-12720][python][docs] Add the Python Table API Sphinx docs
d70e8e2 is described below
commit d70e8e2e92b2174a313542a3485f4a5e2681314a
Author: Huang Xingbo <hx...@gmail.com>
AuthorDate: Thu Jun 13 20:40:30 2019 +0800
[FLINK-12720][python][docs] Add the Python Table API Sphinx docs
This closes #8774
---
.gitignore | 1 +
docs/_config.yml | 1 +
docs/_includes/sidenav.html | 1 +
flink-python/dev/lint-python.sh | 60 +++++-
flink-python/docs/Makefile | 137 +++++++++++++
flink-python/docs/_static/pyflink.js | 116 +++++++++++
flink-python/docs/_templates/layout.html | 22 +++
flink-python/docs/conf.py | 213 +++++++++++++++++++++
flink-python/docs/index.rst | 39 ++++
flink-python/docs/pyflink.rst | 33 ++++
flink-python/docs/pyflink.table.rst | 46 +++++
flink-python/pyflink/table/__init__.py | 80 ++++----
.../table/{table_descriptor.py => descriptors.py} | 87 +++++----
flink-python/pyflink/table/query_config.py | 9 +-
.../pyflink/table/{table_sink.py => sinks.py} | 0
.../pyflink/table/{table_source.py => sources.py} | 0
flink-python/pyflink/table/table.py | 135 ++++++++-----
flink-python/pyflink/table/table_environment.py | 116 +++++++----
.../table/tests/test_catalog_completeness.py | 2 +-
.../pyflink/table/tests/test_descriptor.py | 4 +-
flink-python/pyflink/table/tests/test_window.py | 5 +-
flink-python/pyflink/table/types.py | 63 +++---
flink-python/pyflink/table/window.py | 4 +
flink-python/pyflink/testing/test_case_utils.py | 2 +-
24 files changed, 966 insertions(+), 210 deletions(-)
diff --git a/.gitignore b/.gitignore
index d664bee..876a676 100644
--- a/.gitignore
+++ b/.gitignore
@@ -28,6 +28,7 @@ flink-runtime-web/web-dashboard/web/
flink-python/dist/
flink-python/build/
flink-python/pyflink.egg-info/
+flink-python/docs/_build
flink-python/.tox/
flink-python/dev/download
flink-python/dev/.conda/
diff --git a/docs/_config.yml b/docs/_config.yml
index 48ec910..c3c91b2 100644
--- a/docs/_config.yml
+++ b/docs/_config.yml
@@ -51,6 +51,7 @@ baseurl: //ci.apache.org/projects/flink/flink-docs-master
stable_baseurl: //ci.apache.org/projects/flink/flink-docs-stable
javadocs_baseurl: //ci.apache.org/projects/flink/flink-docs-master
+pythondocs_baseurl: //ci.apache.org/projects/flink/flink-docs-master
# Flag whether this is a stable version or not. Used for the quickstart page.
is_stable: false
diff --git a/docs/_includes/sidenav.html b/docs/_includes/sidenav.html
index 7f200b1..73edab1 100644
--- a/docs/_includes/sidenav.html
+++ b/docs/_includes/sidenav.html
@@ -137,6 +137,7 @@ level is determined by 'nav-pos'.
<li class="divider"></li>
<li><a href="{{ site.javadocs_baseurl }}/api/java"><i class="fa fa-external-link title" aria-hidden="true"></i> Javadocs</a></li>
<li><a href="{{ site.javadocs_baseurl }}/api/scala/index.html#org.apache.flink.api.scala.package"><i class="fa fa-external-link title" aria-hidden="true"></i> Scaladocs</a></li>
+ <li><a href="{{ site.pythondocs_baseurl }}/api/python"><i class="fa fa-external-link title" aria-hidden="true"></i> Pythondocs</a></li>
<li><a href="http://flink.apache.org"><i class="fa fa-external-link title" aria-hidden="true"></i> Project Page</a></li>
</ul>
diff --git a/flink-python/dev/lint-python.sh b/flink-python/dev/lint-python.sh
index b6db4be..c3a8339 100755
--- a/flink-python/dev/lint-python.sh
+++ b/flink-python/dev/lint-python.sh
@@ -244,6 +244,28 @@ function install_flake8() {
fi
}
+# Install sphinx.
+# In some situations,you need to run the script with "sudo". e.g. sudo ./lint-python.sh
+function install_sphinx() {
+ if [ -f "$SPHINX_PATH" ]; then
+ ${CONDA_PATH} remove sphinx -y -q 2>&1 >/dev/null
+ if [ $? -ne 0 ]; then
+ echo "conda remove sphinx failed \
+ please try to exec the script again.\
+ if failed many times, you can try to exec in the form of sudo ./lint-python.sh -f"
+ exit 1
+ fi
+ fi
+
+ ${CONDA_PATH} install -c anaconda sphinx -y -q 2>&1 >/dev/null
+ if [ $? -ne 0 ]; then
+ echo "conda install sphinx failed \
+ please try to exec the script again.\
+ if failed many times, you can try to exec in the form of sudo ./lint-python.sh -f"
+ exit 1
+ fi
+}
+
# In this function, the script will prepare all kinds of python environments and checks.
function install_environment() {
@@ -304,6 +326,15 @@ function install_environment() {
fi
print_function "STEP" "install flake8... [SUCCESS]"
+ # step-6 install sphinx
+ print_function "STEP" "installing sphinx..."
+ if [ $STEP -lt 6 ]; then
+ install_sphinx
+ STEP=6
+ checkpoint_stage $STAGE $STEP
+ fi
+ print_function "STEP" "install sphinx... [SUCCESS]"
+
print_function "STAGE" "install environment... [SUCCESS]"
}
@@ -467,6 +498,28 @@ function flake8_check() {
print_function "STAGE" "flake8 checks... [SUCCESS]"
fi
}
+
+# Sphinx check
+function sphinx_check() {
+ export SPHINXBUILD=$SPHINX_PATH
+ # cd to $FLINK_PYTHON_DIR
+ pushd "$FLINK_PYTHON_DIR"/docs &> /dev/null
+ make clean
+
+ # the return value of a pipeline is the status of the last command to exit
+ # with a non-zero status or zero if no command exited with a non-zero status
+ set -o pipefail
+ (SPHINXOPTS="-a -W" make html) 2>&1 | tee -a $LOG_FILE
+
+ SPHINXBUILD_STATUS=$?
+ if [ $SPHINXBUILD_STATUS -ne 0 ]; then
+ print_function "STAGE" "sphinx checks... [FAILED]"
+ # Stop the running script.
+ exit 1;
+ else
+ print_function "STAGE" "sphinx checks... [SUCCESS]"
+ fi
+}
###############################################################All Checks Definitions###############################################################
# CURRENT_DIR is "flink/flink-python/dev/"
@@ -485,6 +538,9 @@ TOX_PATH=$CURRENT_DIR/.conda/bin/tox
# flake8 path
FLAKE8_PATH=$CURRENT_DIR/.conda/bin/flake8
+# sphinx path
+SPHINX_PATH=$CURRENT_DIR/.conda/bin/sphinx-build
+
_OLD_PATH="$PATH"
SUPPORT_OS=("Darwin" "Linux")
@@ -515,7 +571,7 @@ echo >$LOG_FILE
CONDA_INSTALL_SH=$CURRENT_DIR/download/miniconda.sh
# stage "install" includes the num of steps.
-STAGE_INSTALL_STEPS=5
+STAGE_INSTALL_STEPS=6
# whether force to restart the script.
FORCE_START=0
@@ -550,7 +606,7 @@ while getopts "hfi:e:l" arg; do
EXCLUDE_CHECKS=($(echo $OPTARG | tr ',' ' ' ))
;;
i)
- SELECT_CHECKS=($(echo $OPTARG | tr ',' ' ' ))
+ INCLUDE_CHECKS=($(echo $OPTARG | tr ',' ' ' ))
;;
l)
printf "current supported checks includes:\n"
diff --git a/flink-python/docs/Makefile b/flink-python/docs/Makefile
new file mode 100644
index 0000000..8584c73
--- /dev/null
+++ b/flink-python/docs/Makefile
@@ -0,0 +1,137 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+ifndef SPHINXBUILD
+ifndef SPHINXPYTHON
+SPHINXBUILD = sphinx-build
+SPHINXPYTHON = python
+endif
+endif
+
+# You can set SPHINXBUILD to specify the sphinx-build or SPHINXPYTHON to specify the python env to call the sphinx module
+# If SPHINXBUILD is set, the command is 'sphinx-build'.
+# Elseif SPHINXPYTHON is set, the command is 'python -m sphinx'.
+ifdef SPHINXBUILD
+# Check whether sphinx-build is installed
+ifeq ($(shell hash $(SPHINXBUILD) >/dev/null 2>&1; echo $$?), 1)
+$(error The command '$(SPHINXBUILD)' was not found.If you don't have Sphinx installed, grab it from http://sphinx-doc.org/)
+endif
+else
+ifeq ($(shell $(SPHINXPYTHON) -c 'import sphinx' >/dev/null 2>&1; echo $$?), 1)
+# Check whether sphinx module is installed in $(SPHINXPYTHON)
+$(error '$(SPHINXPYTHON)' doesn't install sphinx module.You can use specified pip of $(SPHINXPYTHON) to install sphinx)
+endif
+SPHINXBUILD = $(SPHINXPYTHON) -m sphinx
+endif
+
+# The project depends on py4j, so py4j-*-src.zip should be set in PYTHONPATH.
+py4j_lib = $(shell echo ../lib/py4j-*-src.zip)
+export PYTHONPATH = $(realpath $(py4j_lib))
+
+# You can set these variables from the command line.
+SPHINXOPTS ?=
+PAPER ?=
+SOURCEDIR ?= .
+BUILDDIR ?= _build
+
+# Internal variables.
+PAPEROPT_a4 = -D latex_paper_size=a4
+PAPEROPT_letter = -D latex_paper_size=letter
+ALLSPHINXOPTS = -d _build/doctrees $(PAPEROPT_$(PAPER)) $(SPHINXOPTS) .
+
+.PHONY: help clean html dirhtml pickle json htmlhelp qthelp latex changes linkcheck doctest
+
+help:
+ @echo "Please use \`make <target>' where <target> is one of"
+ @echo " html to make standalone HTML files"
+ @echo " dirhtml to make HTML files named index.html in directories"
+ @echo " pickle to make pickle files"
+ @echo " json to make JSON files"
+ @echo " htmlhelp to make HTML files and a HTML help project"
+ @echo " qthelp to make HTML files and a qthelp project"
+ @echo " latex to make LaTeX files, you can set PAPER=a4 or PAPER=letter"
+ @echo " changes to make an overview of all changed/added/deprecated items"
+ @echo " linkcheck to check all external links for integrity"
+ @echo " doctest to run all doctests embedded in the documentation (if enabled)"
+
+clean:
+ -rm -rf _build/*
+
+html:
+ $(SPHINXBUILD) -b html $(ALLSPHINXOPTS) _build/html
+ @echo
+ @echo "Build finished. The HTML pages are in _build/html."
+
+dirhtml:
+ $(SPHINXBUILD) -b dirhtml $(ALLSPHINXOPTS) _build/dirhtml
+ @echo
+ @echo "Build finished. The HTML pages are in _build/dirhtml."
+
+pickle:
+ $(SPHINXBUILD) -b pickle $(ALLSPHINXOPTS) _build/pickle
+ @echo
+ @echo "Build finished; now you can process the pickle files."
+
+json:
+ $(SPHINXBUILD) -b json $(ALLSPHINXOPTS) _build/json
+ @echo
+ @echo "Build finished; now you can process the JSON files."
+
+htmlhelp:
+ $(SPHINXBUILD) -b htmlhelp $(ALLSPHINXOPTS) _build/htmlhelp
+ @echo
+ @echo "Build finished; now you can run HTML Help Workshop with the" \
+ ".hhp project file in _build/htmlhelp."
+
+qthelp:
+ $(SPHINXBUILD) -b qthelp $(ALLSPHINXOPTS) _build/qthelp
+ @echo
+ @echo "Build finished; now you can run "qcollectiongenerator" with the" \
+ ".qhcp project file in _build/qthelp, like this:"
+ @echo "# qcollectiongenerator _build/qthelp/pyflinkdoc.qhcp"
+ @echo "To view the help file:"
+ @echo "# assistant -collectionFile _build/qthelp/pyflinkdoc.qhc"
+
+latex:
+ $(SPHINXBUILD) -b latex $(ALLSPHINXOPTS) _build/latex
+ @echo
+ @echo "Build finished; the LaTeX files are in _build/latex."
+ @echo "Run \`make all-pdf' or \`make all-ps' in that directory to" \
+ "run these through (pdf)latex."
+
+changes:
+ $(SPHINXBUILD) -b changes $(ALLSPHINXOPTS) _build/changes
+ @echo
+ @echo "The overview file is in _build/changes."
+
+linkcheck:
+ $(SPHINXBUILD) -b linkcheck $(ALLSPHINXOPTS) _build/linkcheck
+ @echo
+ @echo "Link check complete; look for any errors in the above output " \
+ "or in _build/linkcheck/output.txt."
+
+doctest:
+ $(SPHINXBUILD) -b doctest $(ALLSPHINXOPTS) _build/doctest
+ @echo "Testing of doctests in the sources finished, look at the " \
+ "results in _build/doctest/output.txt."
+
+pdf: latex
+ cd _build/latex && make all-pdf
+
+pdf-html: pdf html
+ cp _build/latex/pyflinkdoc.pdf _build/html
diff --git a/flink-python/docs/_static/pyflink.js b/flink-python/docs/_static/pyflink.js
new file mode 100644
index 0000000..113ae1f
--- /dev/null
+++ b/flink-python/docs/_static/pyflink.js
@@ -0,0 +1,116 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+$(function () {
+
+ function startsWith(s, prefix) {
+ return s && s.indexOf(prefix) === 0;
+ }
+
+ // Constructs map stores modules
+ function buildModuleLinkMap() {
+ var moduleLinkMap = {};
+ $('div.sphinxsidebar a.reference.internal').each(function (_, a) {
+ var href = $(a).attr('href');
+ // Whether is a module.
+ if (startsWith(href, '#module-')) {
+ var moduleName = href.substr(8);
+ moduleLinkMap[moduleName] = [$(a), null];
+ }
+ });
+ return moduleLinkMap;
+ }
+
+ function getDivNoteMap(dd) {
+ //Stores the divs that includes 'Special note' such as 'Deprecated',Experimental.
+ var divNoteMap = {};
+ // <div class="admonition note">
+ // <p class="first admonition-title">Note</p>
+ // <p class="last">WARN: Flink Internal Use Only</p>
+ // </div>
+ // The structure will get 'WARN: Flink Internal Use Only'
+ dd.find('> div.admonition.note > p.last').each(function (_, p) {
+ var text = $(p).text();
+ // Whether is Deprecated.
+ if (!divNoteMap.deprecated && startsWith(text, 'Deprecated')) {
+ divNoteMap.deprecated = $(p).parent();
+ }
+ // Whether is Experimental.
+ if (!divNoteMap.experimental && startsWith(text, 'Experimental')) {
+ divNoteMap.experimental = $(p).parent();
+ }
+ });
+ return divNoteMap;
+ }
+
+ function getPackageName(name) {
+ var last_idx = name.lastIndexOf('.');
+ return last_idx === -1 ? '' : name.substr(0, last_idx);
+ }
+
+ function buildTag(text, cls, tooltip) {
+ return '<span class="pys-tag ' + cls + ' hasTooltip">' + text + '<span class="tooltip">'
+ + tooltip + '</span></span>'
+ }
+
+
+ var moduleLinkMap = buildModuleLinkMap();
+
+ // The class field of dl label includes 'class','method','classmethod'.
+ $('dl.class, dl.function').each(function (_, dl) {
+
+ dl = $(dl);
+ var dt = dl.children('dt').eq(0);
+ var dd = dl.children('dd').eq(0);
+
+ // Name of the class(method,classmethod).
+ // e.g. <dt id="pyflink.table.TableEnvironment"> will get 'pyflink.table.TableEnvironment'
+ var name = dt.attr('id');
+ // Desc of the class(method,classmethod).
+ // e.g. <code class="descname">TableEnvironment</code> will get 'TableEnvironment'
+ var descName = dt.find('> .descname').text();
+ var divNoteMap = getDivNoteMap(dd);
+
+ if (name) {
+ // e.g. 'pyflink.table.TableEnvironment' will get 'pyflink.table'
+ var packageName = getPackageName(name);
+
+ var moduleLink = moduleLinkMap[packageName];
+ if (moduleLink) {
+ if (moduleLink[1] === null) {
+ moduleLink[1] = $('<ul/>');
+ moduleLink[0].parent().append(moduleLink[1]);
+ }
+ var tags = '';
+ if (divNoteMap.experimental) {
+ tags += buildTag('E', 'pys-tag-experimental', 'Experimental');
+ divNoteMap.experimental.addClass('pys-note pys-note-experimental');
+ }
+ if (divNoteMap.deprecated) {
+ tags += buildTag('D', 'pys-tag-deprecated', 'Deprecated');
+ divNoteMap.deprecated.addClass('pys-note pys-note-deprecated');
+ }
+ var li = $('<li/>');
+ var a = $('<a href="#' + name + '">' + descName + '</a>');
+ li.append(a);
+ li.append(tags);
+ moduleLink[1].append(li);
+ moduleLinkMap[name] = [a, null];
+ }
+ }
+ });
+});
diff --git a/flink-python/docs/_templates/layout.html b/flink-python/docs/_templates/layout.html
new file mode 100644
index 0000000..68848df
--- /dev/null
+++ b/flink-python/docs/_templates/layout.html
@@ -0,0 +1,22 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+{% extends "!layout.html" %}
+{% set script_files = script_files + ["_static/pyflink.js"] %}
+{% block rootrellink %}
+{% endblock %}
diff --git a/flink-python/docs/conf.py b/flink-python/docs/conf.py
new file mode 100644
index 0000000..598ba6b
--- /dev/null
+++ b/flink-python/docs/conf.py
@@ -0,0 +1,213 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from __future__ import print_function
+import os
+import sys
+
+# -- Path setup --------------------------------------------------------------
+
+# If extensions (or modules to document with autodoc) are in another directory,
+# add these directories to sys.path here. If the directory is relative to the
+# documentation root, use os.path.abspath to make it absolute, like shown here.
+sys.path.insert(0, os.path.abspath('.'))
+sys.path.insert(0, os.path.abspath('..'))
+
+# -- Project information -----------------------------------------------------
+
+# project = u'Flink Python Table API'
+project = u'PyFlink'
+copyright = u''
+author = u'Author'
+
+version_file = os.path.join("..", 'pyflink/version.py')
+try:
+ exec(open(version_file).read())
+except IOError:
+ print("Failed to load PyFlink version file for packaging. " +
+ "'%s' not found!" % version_file,
+ file=sys.stderr)
+ sys.exit(-1)
+# The short X.Y version
+version = __version__ # noqa
+# The full version, including alpha/beta/rc tags
+release = os.environ.get('RELEASE_VERSION', version)
+
+# -- General configuration ---------------------------------------------------
+
+# If your documentation needs a minimal Sphinx version, state it here.
+#
+# needs_sphinx = '1.0'
+
+# Add any Sphinx extension module names here, as strings. They can be
+# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
+# ones.
+extensions = [
+ 'sphinx.ext.autodoc',
+ 'sphinx.ext.viewcode',
+ 'sphinx.ext.mathjax',
+ 'sphinx.ext.inheritance_diagram',
+ 'sphinx.ext.doctest',
+]
+
+# Add any paths that contain templates here, relative to this directory.
+templates_path = ['_templates']
+
+# The suffix(es) of source filenames.
+# You can specify multiple suffix as a list of string:
+#
+# source_suffix = ['.rst', '.md']
+source_suffix = '.rst'
+
+# The master toctree document.
+master_doc = 'index'
+
+# The language for content autogenerated by Sphinx. Refer to documentation
+# for a list of supported languages.
+#
+# This is also used if you do content translation via gettext catalogs.
+# Usually you set "language" from the command line for these cases.
+language = None
+
+# List of patterns, relative to source directory, that match files and
+# directories to ignore when looking for source files.
+# This pattern also affects html_static_path and html_extra_path.
+exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
+
+# The name of the Pygments (syntax highlighting) style to use.
+pygments_style = 'sphinx'
+
+# Look at the first line of the docstring for function and method signatures.
+autodoc_docstring_signature = True
+
+# -- Options for HTML output -------------------------------------------------
+
+# The theme to use for HTML and HTML Help pages. See the documentation for
+# a list of builtin themes.
+#
+html_theme = 'nature'
+
+# Theme options are theme-specific and customize the look and feel of a theme
+# further. For a list of options available for each theme, see the
+# documentation.
+#
+# html_theme_options = {}
+
+# The name of an image file (relative to this directory) to place at the top
+# of the sidebar.
+html_logo = "../../docs/page/img/navbar-brand-logo.jpg"
+
+# The name of an image file (within the static path) to use as favicon of the
+# docs. This file should be a Windows icon file (.ico) being 16x16 or 32x32
+# pixels large.
+html_favicon = html_logo
+
+# If false, no module index is generated.
+html_domain_indices = False
+
+# If false, no index is generated.
+html_use_index = False
+# Add any paths that contain custom static files (such as style sheets) here,
+# relative to this directory. They are copied after the builtin static files,
+# so a file named "default.css" will overwrite the builtin "default.css".
+html_static_path = ['_static']
+
+# Custom sidebar templates, must be a dictionary that maps document names
+# to template names.
+#
+# The default sidebars (for documents that don't match any pattern) are
+# defined by theme itself. Builtin themes are using these templates by
+# default: ``['localtoc.html', 'relations.html', 'sourcelink.html',
+# 'searchbox.html']``.
+#
+# html_sidebars = {}
+
+# -- Options for HTMLHelp output ---------------------------------------------
+
+# Output file base name for HTML help builder.
+htmlhelp_basename = 'pyflinkdoc'
+
+# -- Options for LaTeX output ------------------------------------------------
+
+latex_elements = {
+ # The paper size ('letterpaper' or 'a4paper').
+ #
+ # 'papersize': 'letterpaper',
+
+ # The font size ('10pt', '11pt' or '12pt').
+ #
+ # 'pointsize': '10pt',
+
+ # Additional stuff for the LaTeX preamble.
+ #
+ # 'preamble': '',
+
+ # Latex figure (float) alignment
+ #
+ # 'figure_align': 'htbp',
+}
+
+# Grouping the document tree into LaTeX files. List of tuples
+# (source start file, target name, title,
+# author, documentclass [howto, manual, or own class]).
+latex_documents = [
+ (master_doc, 'pyflink.tex', u'pyflink Documentation',
+ [author], 'manual'),
+]
+
+# -- Options for manual page output ------------------------------------------
+
+# One entry per manual page. List of tuples
+# (source start file, name, description, authors, manual section).
+man_pages = [
+ (master_doc, 'pyflink', u'pyflink Documentation',
+ [author], 1)
+]
+
+# -- Options for Texinfo output ----------------------------------------------
+
+# Grouping the document tree into Texinfo files. List of tuples
+# (source start file, target name, title, author,
+# dir menu entry, description, category)
+texinfo_documents = [
+ (master_doc, 'pyflink', u'pyflink Documentation',
+ author, 'pyflink', 'One line description of project.',
+ 'Miscellaneous'),
+]
+
+# -- Options for Epub output -------------------------------------------------
+
+# Bibliographic Dublin Core info.
+epub_title = u'pyflink'
+epub_author = u'Author'
+epub_publisher = u'Author'
+epub_copyright = u'2019, Author'
+
+# The unique identifier of the text. This can be a ISBN number
+# or the project homepage.
+#
+# epub_identifier = ''
+
+# A unique identification for the text.
+#
+# epub_uid = ''
+
+# A list of files that should not be packed into the epub file.
+epub_exclude_files = ['search.html']
+
+# -- Extension configuration -------------------------------------------------
diff --git a/flink-python/docs/index.rst b/flink-python/docs/index.rst
new file mode 100644
index 0000000..53fef00
--- /dev/null
+++ b/flink-python/docs/index.rst
@@ -0,0 +1,39 @@
+.. ################################################################################
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ ################################################################################
+
+Welcome to Flink Python API Docs!
+==================================================
+
+.. toctree::
+ :maxdepth: 2
+ :caption: Contents
+
+ pyflink
+ pyflink.table
+
+
+Core Classes:
+---------------
+
+ :class:`pyflink.table.TableEnvironment`
+
+ Main entry point for Flink Table functionality.
+
+ :class:`pyflink.table.Table`
+
+ Core component of the Flink Table API. The Flink Table API is built around :class:`~pyflink.table.Table`.
diff --git a/flink-python/docs/pyflink.rst b/flink-python/docs/pyflink.rst
new file mode 100644
index 0000000..fb241f7
--- /dev/null
+++ b/flink-python/docs/pyflink.rst
@@ -0,0 +1,33 @@
+.. ################################################################################
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ ################################################################################
+
+pyflink package
+===============
+
+Subpackages
+-----------
+
+.. toctree::
+ :maxdepth: 1
+
+ pyflink.table
+
+.. automodule:: pyflink
+ :members:
+ :undoc-members:
+ :show-inheritance:
diff --git a/flink-python/docs/pyflink.table.rst b/flink-python/docs/pyflink.table.rst
new file mode 100644
index 0000000..af69909
--- /dev/null
+++ b/flink-python/docs/pyflink.table.rst
@@ -0,0 +1,46 @@
+.. ################################################################################
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ ################################################################################
+
+pyflink.table package
+=====================
+
+Module contents
+---------------
+
+.. automodule:: pyflink.table
+ :members:
+ :undoc-members:
+ :show-inheritance:
+
+pyflink.table.window module
+---------------------------
+.. automodule:: pyflink.table.window
+ :members:
+ :undoc-members:
+
+pyflink.table.descriptors module
+-------------------------------------
+.. automodule:: pyflink.table.descriptors
+ :members:
+ :undoc-members:
+
+pyflink.table.catalog module
+----------------------------
+.. automodule:: pyflink.table.catalog
+ :members:
+ :undoc-members:
diff --git a/flink-python/pyflink/table/__init__.py b/flink-python/pyflink/table/__init__.py
index c909270..db352df 100644
--- a/flink-python/pyflink/table/__init__.py
+++ b/flink-python/pyflink/table/__init__.py
@@ -27,70 +27,58 @@ Important classes of Flink Table API:
- :class:`pyflink.table.TableConfig`
A config to define the runtime behavior of the Table API.
It is necessary when creating :class:`TableEnvironment`.
+ - :class:`pyflink.table.StreamQueryConfig` and :class:`pyflink.table.BatchQueryConfig`
+ A query config holds parameters to configure the behavior of queries.
- :class:`pyflink.table.TableSource`
Defines an external data source as a table.
- :class:`pyflink.table.TableSink`
Specifies how to emit a table to an external system or location.
+ - :class:`pyflink.table.DataTypes`
+ Defines a list of data types available.
+ - :class:`pyflink.table.Row`
+ A row in a :class:`Table`.
+ - :class:`pyflink.table.window`
+ Helper classes for working with :class:`pyflink.table.window.GroupWindow`
+ (:class:`pyflink.table.window.Tumble`, :class:`pyflink.table.window.Session`,
+ :class:`pyflink.table.window.Slide`) and :class:`pyflink.table.window.OverWindow` window
+ (:class:`pyflink.table.window.Over`).
+ - :class:`pyflink.table.descriptors`
+ Helper classes that describes DDL information, such as how to connect to another system,
+ the format of data, the schema of table, the event time attribute in the schema, etc.
+ - :class:`pyflink.table.catalog`
+ Responsible for reading and writing metadata such as database/table/views/UDFs
+ from a registered :class:`pyflink.table.catalog.Catalog`.
+ - :class:`pyflink.table.TableSchema`
+ Represents a table's structure with field names and data types.
"""
-from pyflink.table.catalog import (Catalog, CatalogDatabase, CatalogBaseTable, CatalogPartition,
- CatalogFunction, ObjectPath, CatalogPartitionSpec,
- CatalogTableStatistics, CatalogColumnStatistics, HiveCatalog,
- HiveCatalogDatabase, HiveCatalogFunction, HiveCatalogPartition,
- HiveCatalogTable, HiveCatalogView)
-from pyflink.table.table import Table
+from __future__ import absolute_import
+
+from pyflink.table.query_config import BatchQueryConfig, StreamQueryConfig
+from pyflink.table.table import Table, GroupedTable, GroupWindowedTable, OverWindowedTable, \
+ WindowGroupedTable
from pyflink.table.table_config import TableConfig
-from pyflink.table.table_environment import (TableEnvironment, StreamTableEnvironment,
- BatchTableEnvironment)
-from pyflink.table.table_sink import TableSink, CsvTableSink
-from pyflink.table.table_source import TableSource, CsvTableSource
+from pyflink.table.table_environment import TableEnvironment
+from pyflink.table.sinks import TableSink, CsvTableSink
+from pyflink.table.sources import TableSource, CsvTableSource
from pyflink.table.types import DataTypes, UserDefinedType, Row
-from pyflink.table.window import Tumble, Session, Slide, Over
-from pyflink.table.table_descriptor import (Rowtime, Schema, OldCsv, Csv, Avro, Json, FileSystem,
- Kafka, Elasticsearch)
from pyflink.table.table_schema import TableSchema
__all__ = [
'TableEnvironment',
- 'StreamTableEnvironment',
- 'BatchTableEnvironment',
'Table',
+ 'GroupedTable',
+ 'GroupWindowedTable',
+ 'OverWindowedTable',
+ 'WindowGroupedTable',
'TableConfig',
+ 'StreamQueryConfig',
+ 'BatchQueryConfig',
'TableSink',
'TableSource',
'CsvTableSink',
'CsvTableSource',
'DataTypes',
- 'Tumble',
- 'Session',
- 'Slide',
- 'Over',
- 'Rowtime',
- 'Schema',
- 'OldCsv',
- 'Csv',
- 'Avro',
- 'Json',
- 'FileSystem',
'UserDefinedType',
'Row',
- 'Kafka',
- 'Elasticsearch',
- 'TableSchema',
-
- # Catalog APIs
- 'Catalog',
- 'CatalogDatabase',
- 'CatalogBaseTable',
- 'CatalogPartition',
- 'CatalogFunction',
- 'ObjectPath',
- 'CatalogPartitionSpec',
- 'CatalogTableStatistics',
- 'CatalogColumnStatistics',
- 'HiveCatalog',
- 'HiveCatalogDatabase',
- 'HiveCatalogFunction',
- 'HiveCatalogPartition',
- 'HiveCatalogTable',
- 'HiveCatalogView'
+ 'TableSchema'
]
diff --git a/flink-python/pyflink/table/table_descriptor.py b/flink-python/pyflink/table/descriptors.py
similarity index 95%
rename from flink-python/pyflink/table/table_descriptor.py
rename to flink-python/pyflink/table/descriptors.py
index 9570a31..d31561c 100644
--- a/flink-python/pyflink/table/table_descriptor.py
+++ b/flink-python/pyflink/table/descriptors.py
@@ -36,7 +36,10 @@ __all__ = [
'Csv',
'Avro',
'Json',
- 'FileSystem'
+ 'FileSystem',
+ 'ConnectTableDescriptor',
+ 'StreamTableDescriptor',
+ 'BatchTableDescriptor'
]
@@ -95,7 +98,8 @@ class Rowtime(Descriptor):
API record into the rowtime attribute and thus preserves the assigned timestamps from the
source.
- ..note::
+ .. note::
+
This extractor only works in streaming environments.
:return: This rowtime descriptor.
@@ -174,7 +178,8 @@ class Schema(Descriptor):
"""
Describes a schema of a table.
- ..note::
+ .. note::
+
Field names are matched by the exact name by default (case sensitive).
"""
@@ -188,7 +193,7 @@ class Schema(Descriptor):
Sets the schema with field names and the types. Required.
This method overwrites existing fields added with
- :func:`~pyflink.table.table_descriptor.Schema.field`.
+ :func:`~pyflink.table.descriptors.Schema.field`.
:param table_schema: The :class:`TableSchema` object.
:return: This schema object.
@@ -220,7 +225,8 @@ class Schema(Descriptor):
E.g. field("myString", Types.STRING).from_origin_field("CSV_MY_STRING")
- ..note::
+ .. note::
+
Field names are matched by the exact name by default (case sensitive).
:param origin_field_name: The origin field name.
@@ -269,13 +275,15 @@ class OldCsv(FormatDescriptor):
"""
Format descriptor for comma-separated values (CSV).
- ..note::
+ .. note::
+
This descriptor describes Flink's non-standard CSV table source/sink. In the future, the
descriptor will be replaced by a proper RFC-compliant version. Use the RFC-compliant `Csv`
format in the dedicated `flink-formats/flink-csv` module instead when writing to Kafka. Use
the old one for stream/batch filesystem operations for now.
- ..note::
+ .. note::
+
Deprecated: use the RFC-compliant `Csv` format instead when writing to Kafka.
"""
@@ -295,8 +303,8 @@ class OldCsv(FormatDescriptor):
return self
def line_delimiter(self, delimiter):
- """
- Sets the line delimiter, "\n" by default.
+ r"""
+ Sets the line delimiter, "\\n" by default.
:param delimiter: The line delimiter.
:return: This :class:`OldCsv` object.
@@ -309,10 +317,10 @@ class OldCsv(FormatDescriptor):
Sets the schema with field names and the types. Required.
This method overwrites existing fields added with
- :func:`~pyflink.table.table_descriptor.OldCsv.field`.
+ :func:`~pyflink.table.descriptors.OldCsv.field`.
:param table_schema: The :class:`TableSchema` object.
- :return: This schema object.
+ :return: This :class:`OldCsv` object.
"""
self._j_csv = self._j_csv.schema(table_schema._j_table_schema)
return self
@@ -379,7 +387,8 @@ class Csv(FormatDescriptor):
This descriptor aims to comply with RFC-4180 ("Common Format and MIME Type for
Comma-Separated Values (CSV) Files") proposed by the Internet Engineering Task Force (IETF).
- ..note::
+ .. note::
+
This descriptor does not describe Flink's old non-standard CSV table
source/sink. Currently, this descriptor can be used when writing to Kafka. The old one is
still available as :class:`OldCsv` for stream/batch filesystem operations.
@@ -403,8 +412,8 @@ class Csv(FormatDescriptor):
return self
def line_delimiter(self, delimiter):
- """
- Sets the line delimiter ("\n" by default; otherwise "\r" or "\r\n" are allowed).
+ r"""
+ Sets the line delimiter ("\\n" by default; otherwise "\\r" or "\\r\\n" are allowed).
:param delimiter: The line delimiter.
:return: This :class:`Csv` object.
@@ -747,7 +756,7 @@ class Kafka(ConnectorDescriptor):
If the provided map of offsets contains entries whose partition is not subscribed by the
consumer, the entry will be ignored. If the consumer subscribes to a partition that does
not exist in the provided map of offsets, the consumer will fallback to the default group
- offset behaviour(see :func:`pyflink.table.table_descriptor.Kafka.start_from_group_offsets`)
+ offset behaviour(see :func:`pyflink.table.descriptors.Kafka.start_from_group_offsets`)
for that particular partition.
If the specified offset for a partition is invalid, or the behaviour for that partition is
@@ -772,10 +781,10 @@ class Kafka(ConnectorDescriptor):
Configures to start reading partitions from specific offsets and specifies the given offset
for the given partition.
- see :func:`pyflink.table.table_descriptor.Kafka.start_from_specific_offsets`
+ see :func:`pyflink.table.descriptors.Kafka.start_from_specific_offsets`
- :param partition:
- :param specific_offset:
+ :param partition: Partition id.
+ :param specific_offset: Specified offset in given partition.
:return: This object.
"""
self._j_kafka = self._j_kafka.startFromSpecificOffset(int(partition), int(specific_offset))
@@ -787,7 +796,8 @@ class Kafka(ConnectorDescriptor):
This strategy ensures that each Flink partition ends up in one Kafka partition.
- ..note::
+ .. note::
+
One Kafka partition can contain multiple Flink partitions. Examples:
More Flink partitions than Kafka partitions. Some (or all) Kafka partitions contain
@@ -820,7 +830,8 @@ class Kafka(ConnectorDescriptor):
This strategy ensures that records will be distributed to Kafka partitions in a
round-robin fashion.
- ..note::
+ .. note::
+
This strategy is useful to avoid an unbalanced partitioning. However, it will cause a
lot of network connections between all the Flink instances and all the Kafka brokers.
@@ -980,7 +991,8 @@ class Elasticsearch(ConnectorDescriptor):
Disables flushing on checkpoint. When disabled, a sink will not wait for all pending action
requests to be acknowledged by Elasticsearch on checkpoints.
- ..note::
+ .. note::
+
If flushing on checkpoint is disabled, a Elasticsearch sink does NOT
provide any strong guarantees for at-least-once delivery of action requests.
@@ -1061,8 +1073,8 @@ class Elasticsearch(ConnectorDescriptor):
Sets the maximum number of retries for a backoff attempt when flushing bulk requests.
Make sure to enable backoff by selecting a strategy (
- :func:`pyflink.table.table_descriptor.Elasticsearch.bulk_flush_backoff_constant` or
- :func:`pyflink.table.table_descriptor.Elasticsearch.bulk_flush_backoff_exponential`).
+ :func:`pyflink.table.descriptors.Elasticsearch.bulk_flush_backoff_constant` or
+ :func:`pyflink.table.descriptors.Elasticsearch.bulk_flush_backoff_exponential`).
:param max_retries: The maximum number of retries.
:return: This object.
@@ -1079,8 +1091,8 @@ class Elasticsearch(ConnectorDescriptor):
(in milliseconds).
Make sure to enable backoff by selecting a strategy (
- :func:`pyflink.table.table_descriptor.Elasticsearch.bulk_flush_backoff_constant` or
- :func:`pyflink.table.table_descriptor.Elasticsearch.bulk_flush_backoff_exponential`).
+ :func:`pyflink.table.descriptors.Elasticsearch.bulk_flush_backoff_constant` or
+ :func:`pyflink.table.descriptors.Elasticsearch.bulk_flush_backoff_exponential`).
:param delay: Delay between each backoff attempt (in milliseconds).
:return: This object.
@@ -1121,9 +1133,9 @@ class ConnectTableDescriptor(Descriptor):
__metaclass__ = ABCMeta
- def __init__(self, j_table_descriptor):
- self._j_table_descriptor = j_table_descriptor
- super(ConnectTableDescriptor, self).__init__(self._j_table_descriptor)
+ def __init__(self, j_connect_table_descriptor):
+ self._j_connect_table_descriptor = j_connect_table_descriptor
+ super(ConnectTableDescriptor, self).__init__(self._j_connect_table_descriptor)
def with_format(self, format_descriptor):
"""
@@ -1133,8 +1145,8 @@ class ConnectTableDescriptor(Descriptor):
e.g. :class:`OldCsv`.
:return: This object.
"""
- self._j_table_descriptor = \
- self._j_table_descriptor.withFormat(format_descriptor._j_format_descriptor)
+ self._j_connect_table_descriptor = \
+ self._j_connect_table_descriptor.withFormat(format_descriptor._j_format_descriptor)
return self
def with_schema(self, schema):
@@ -1144,7 +1156,8 @@ class ConnectTableDescriptor(Descriptor):
:type schema: The :class:`Schema` object for the resulting table.
:return: This object.
"""
- self._j_table_descriptor = self._j_table_descriptor.withSchema(schema._j_schema)
+ self._j_connect_table_descriptor = \
+ self._j_connect_table_descriptor.withSchema(schema._j_schema)
return self
def register_table_sink(self, name):
@@ -1155,7 +1168,7 @@ class ConnectTableDescriptor(Descriptor):
:param name: Table name to be registered in the table environment.
:return: This object.
"""
- self._j_table_descriptor = self._j_table_descriptor.registerTableSink(name)
+ self._j_connect_table_descriptor = self._j_connect_table_descriptor.registerTableSink(name)
return self
def register_table_source(self, name):
@@ -1166,7 +1179,8 @@ class ConnectTableDescriptor(Descriptor):
:param name: Table name to be registered in the table environment.
:return: This object.
"""
- self._j_table_descriptor = self._j_table_descriptor.registerTableSource(name)
+ self._j_connect_table_descriptor = \
+ self._j_connect_table_descriptor.registerTableSource(name)
return self
def register_table_source_and_sink(self, name):
@@ -1177,7 +1191,8 @@ class ConnectTableDescriptor(Descriptor):
:param name: Table name to be registered in the table environment.
:return: This object.
"""
- self._j_table_descriptor = self._j_table_descriptor.registerTableSourceAndSink(name)
+ self._j_connect_table_descriptor = \
+ self._j_connect_table_descriptor.registerTableSourceAndSink(name)
return self
@@ -1247,5 +1262,5 @@ class BatchTableDescriptor(ConnectTableDescriptor):
"""
def __init__(self, j_batch_table_descriptor):
- self.j_batch_table_descriptor = j_batch_table_descriptor
- super(BatchTableDescriptor, self).__init__(self.j_batch_table_descriptor)
+ self._j_batch_table_descriptor = j_batch_table_descriptor
+ super(BatchTableDescriptor, self).__init__(self._j_batch_table_descriptor)
diff --git a/flink-python/pyflink/table/query_config.py b/flink-python/pyflink/table/query_config.py
index 9a2411c..d39109a 100644
--- a/flink-python/pyflink/table/query_config.py
+++ b/flink-python/pyflink/table/query_config.py
@@ -56,18 +56,19 @@ class StreamQueryConfig(QueryConfig):
When new data arrives for previously cleaned-up state, the new data will be handled as if it
was the first data. This can result in previous results being overwritten.
- Set to ``datetime.timedelta()``(zero) to never clean-up the state.
+ Set to 0 (zero) to never clean-up the state.
.. note::
+
Cleaning up state requires additional bookkeeping which becomes less expensive for
larger differences of minTime and maxTime. The difference between minTime and maxTime
- must be at least ``datetime.timedelta(minutes=5)``(5 minutes).
+ must be at least 5 minutes.
:param min_time: The minimum time interval for which idle state is retained. Set to
- ``datetime.timedelta()``(zero) to never clean-up the state.
+ 0 (zero) to never clean-up the state.
:param max_time: The maximum time interval for which idle state is retained. Must be at
least 5 minutes greater than minTime. Set to
- ``datetime.timedelta()``(zero) to never clean-up the state.
+ 0 (zero) to never clean-up the state.
:return: :class:`StreamQueryConfig`
"""
# type: (timedelta, timedelta) -> StreamQueryConfig
diff --git a/flink-python/pyflink/table/table_sink.py b/flink-python/pyflink/table/sinks.py
similarity index 100%
rename from flink-python/pyflink/table/table_sink.py
rename to flink-python/pyflink/table/sinks.py
diff --git a/flink-python/pyflink/table/table_source.py b/flink-python/pyflink/table/sources.py
similarity index 100%
rename from flink-python/pyflink/table/table_source.py
rename to flink-python/pyflink/table/sources.py
diff --git a/flink-python/pyflink/table/table.py b/flink-python/pyflink/table/table.py
index aa6d9e5..f3c40bb 100644
--- a/flink-python/pyflink/table/table.py
+++ b/flink-python/pyflink/table/table.py
@@ -27,7 +27,7 @@ from pyflink.util.utils import to_jarray
if sys.version > '3':
xrange = range
-__all__ = ['Table']
+__all__ = ['Table', 'GroupedTable', 'GroupWindowedTable', 'OverWindowedTable', 'WindowGroupedTable']
class Table(object):
@@ -41,6 +41,7 @@ class Table(object):
Example:
::
+
>>> t_config = TableConfig.Builder().as_streaming_execution().set_parallelism(1).build()
>>> t_env = TableEnvironment.create(t_config)
>>> ...
@@ -68,10 +69,11 @@ class Table(object):
Example:
::
+
>>> tab.select("key, value + 'hello'")
:param fields: Expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.select(fields))
@@ -82,10 +84,11 @@ class Table(object):
Example:
::
+
>>> tab.alias("a, b")
:param fields: Field list expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(get_method(self._j_table, "as")(fields))
@@ -96,10 +99,11 @@ class Table(object):
Example:
::
+
>>> tab.filter("name = 'Fred'")
:param predicate: Predicate expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.filter(predicate))
@@ -110,10 +114,11 @@ class Table(object):
Example:
::
+
>>> tab.where("name = 'Fred'")
:param predicate: Predicate expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.where(predicate))
@@ -124,10 +129,11 @@ class Table(object):
Example:
::
+
>>> tab.group_by("key").select("key, value.avg")
:param fields: Group keys.
- :return: The grouped table.
+ :return: The grouped :class:`Table`.
"""
return GroupedTable(self._j_table.groupBy(fields))
@@ -137,30 +143,33 @@ class Table(object):
Example:
::
+
>>> tab.select("key, value").distinct()
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.distinct())
def join(self, right, join_predicate=None):
"""
- Joins two :class:`Table`s. Similar to a SQL join. The fields of the two joined
+ Joins two :class:`Table`. Similar to a SQL join. The fields of the two joined
operations must not overlap, use :func:`~pyflink.table.Table.alias` to rename fields if
necessary. You can use where and select clauses after a join to further specify the
behaviour of the join.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment` .
Example:
::
+
>>> left.join(right).where("a = b && c > 3").select("a, b, d")
>>> left.join(right, "a = b")
:param right: Right table.
:param join_predicate: Optional, the join predicate expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
if join_predicate is not None:
return Table(self._j_table.join(right._j_table, join_predicate))
@@ -169,22 +178,24 @@ class Table(object):
def left_outer_join(self, right, join_predicate=None):
"""
- Joins two :class:`Table`s. Similar to a SQL left outer join. The fields of the two joined
+ Joins two :class:`Table`. Similar to a SQL left outer join. The fields of the two joined
operations must not overlap, use :func:`~pyflink.table.Table.alias` to rename fields if
necessary.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment` and its
:class:`TableConfig` must have null check enabled (default).
Example:
::
+
>>> left.left_outer_join(right).select("a, b, d")
>>> left.left_outer_join(right, "a = b").select("a, b, d")
:param right: Right table.
:param join_predicate: Optional, the join predicate expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
if join_predicate is None:
return Table(self._j_table.leftOuterJoin(right._j_table))
@@ -193,152 +204,168 @@ class Table(object):
def right_outer_join(self, right, join_predicate):
"""
- Joins two :class:`Table`s. Similar to a SQL right outer join. The fields of the two joined
+ Joins two :class:`Table`. Similar to a SQL right outer join. The fields of the two joined
operations must not overlap, use :func:`~pyflink.table.Table.alias` to rename fields if
necessary.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment` and its
:class:`TableConfig` must have null check enabled (default).
Example:
::
+
>>> left.right_outer_join(right, "a = b").select("a, b, d")
:param right: Right table.
:param join_predicate: The join predicate expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.rightOuterJoin(right._j_table, join_predicate))
def full_outer_join(self, right, join_predicate):
"""
- Joins two :class:`Table`s. Similar to a SQL full outer join. The fields of the two joined
+ Joins two :class:`Table`. Similar to a SQL full outer join. The fields of the two joined
operations must not overlap, use :func:`~pyflink.table.Table.alias` to rename fields if
necessary.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment` and its
:class:`TableConfig` must have null check enabled (default).
Example:
::
+
>>> left.full_outer_join(right, "a = b").select("a, b, d")
:param right: Right table.
:param join_predicate: The join predicate expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.fullOuterJoin(right._j_table, join_predicate))
def minus(self, right):
"""
- Minus of two :class:`Table`s with duplicate records removed.
+ Minus of two :class:`Table` with duplicate records removed.
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.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment`.
Example:
::
+
>>> left.minus(right)
:param right: Right table.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.minus(right._j_table))
def minus_all(self, right):
"""
- Minus of two :class:`Table`s. Similar to a SQL EXCEPT ALL.
+ Minus of two :class:`Table`. Similar to a SQL EXCEPT ALL.
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.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment`.
Example:
::
+
>>> left.minus_all(right)
:param right: Right table.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.minusAll(right._j_table))
def union(self, right):
"""
- Unions two :class:`Table`s with duplicate records removed.
+ Unions two :class:`Table` with duplicate records removed.
Similar to a SQL UNION. The fields of the two union operations must fully overlap.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment`.
Example:
::
+
>>> left.union(right)
:param right: Right table.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.union(right._j_table))
def union_all(self, right):
"""
- Unions two :class:`Table`s. Similar to a SQL UNION ALL. The fields of the two union
+ Unions two :class:`Table`. Similar to a SQL UNION ALL. The fields of the two union
operations must fully overlap.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment`.
Example:
::
+
>>> left.union_all(right)
:param right: Right table.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.unionAll(right._j_table))
def intersect(self, right):
"""
- Intersects two :class:`Table`s with duplicate records removed. Intersect returns records
+ Intersects two :class:`Table` with duplicate records removed. 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. Similar to a
SQL INTERSECT. The fields of the two intersect operations must fully overlap.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment`.
Example:
::
+
>>> left.intersect(right)
:param right: Right table.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.intersect(right._j_table))
def intersect_all(self, right):
"""
- Intersects two :class:`Table`s. IntersectAll returns records that exist in both tables.
+ Intersects two :class:`Table`. 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. Similar
to an SQL INTERSECT ALL. The fields of the two intersect operations must fully overlap.
.. note::
+
Both tables must be bound to the same :class:`TableEnvironment`.
Example:
::
+
>>> left.intersect_all(right)
:param right: Right table.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.intersectAll(right._j_table))
@@ -349,10 +376,11 @@ class Table(object):
Example:
::
+
>>> tab.order_by("name.desc")
:param fields: Order fields expression string,
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.orderBy(fields))
@@ -366,13 +394,14 @@ class Table(object):
Example:
::
+
# skips the first 3 rows and returns all following rows.
>>> tab.order_by("name.desc").offset(3)
# skips the first 10 rows and returns the next 5 rows.
>>> tab.order_by("name.desc").offset(10).fetch(5)
:param offset: Number of records to skip.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.offset(offset))
@@ -388,14 +417,16 @@ class Table(object):
Returns the first 3 records.
::
+
>>> tab.order_by("name.desc").fetch(3)
Skips the first 10 rows and returns the next 5 rows.
::
+
>>> tab.order_by("name.desc").offset(10).fetch(5)
:param fetch: The number of records to return. Fetch must be >= 0.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.fetch(fetch))
@@ -413,14 +444,16 @@ class Table(object):
groupBy.
.. note::
+
Computing windowed aggregates on a streaming table is only a parallel operation
if additional grouping attributes are added to the
:func:`~pyflink.table.GroupWindowedTable.group_by` clause.
If the :func:`~pyflink.table.GroupWindowedTable.group_by` only references a GroupWindow
alias, the streamed table will be processed by a single task, i.e., with parallelism 1.
- :param window: A :class:`GroupWindow` created from :class:`Tumble`, :class:`Session` or
- :class:`Slide`.
+ :param window: A :class:`pyflink.table.window.GroupWindow` created from
+ :class:`pyflink.table.window.Tumble`, :class:`pyflink.table.window.Session`
+ or :class:`pyflink.table.window.Slide`.
:return: A :class:`GroupWindowedTable`.
"""
# type: (GroupWindow) -> GroupWindowedTable
@@ -435,16 +468,19 @@ class Table(object):
Example:
::
- >>> table.window(Over.partition_by("c").order_by("rowTime")\
- ... .preceding("10.seconds").alias("ow"))\
+
+ >>> table.window(Over.partition_by("c").order_by("rowTime")\\
+ ... .preceding("10.seconds").alias("ow"))\\
... .select("c, b.count over ow, e.sum over ow")
.. note::
+
Computing over window aggregates on a streaming table is only a parallel
operation if the window is partitioned. Otherwise, the whole stream will be processed
by a single task, i.e., with parallelism 1.
.. note::
+
Over-windows for batch tables are currently not supported.
:param over_windows: :class:`OverWindow`s created from :class:`Over`.
@@ -463,10 +499,11 @@ class Table(object):
Example:
::
+
>>> tab.add_columns("a + 1 as a1, concat(b, 'sunny') as b1")
:param fields: Column list string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.addColumns(fields))
@@ -479,10 +516,11 @@ class Table(object):
Example:
::
+
>>> tab.add_or_replace_columns("a + 1 as a1, concat(b, 'sunny') as b1")
:param fields: Column list string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.addOrReplaceColumns(fields))
@@ -493,10 +531,11 @@ class Table(object):
Example:
::
+
>>> tab.rename_columns("a as a1, b as b1")
:param fields: Column list string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.renameColumns(fields))
@@ -506,10 +545,11 @@ class Table(object):
Example:
::
+
>>> tab.drop_columns("a, b")
:param fields: Column list string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.dropColumns(fields))
@@ -517,10 +557,11 @@ class Table(object):
"""
Writes the :class:`Table` to a :class:`TableSink` that was registered under
the specified name. For the path resolution algorithm see
- :func:`~TableEnvironment.useDatabase`.
+ :func:`~TableEnvironment.use_database`.
Example:
::
+
>>> tab.insert_into("print")
:param table_path: The first part of the path of the registered :class:`TableSink` to which
@@ -566,18 +607,19 @@ class GroupedTable(object):
Example:
::
+
>>> tab.group_by("key").select("key, value.avg + ' The average' as average")
:param fields: Expression string that contains group keys and aggregate function calls.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.select(fields))
class GroupWindowedTable(object):
"""
- A table that has been windowed for :class:`GroupWindow`s.
+ A table that has been windowed for :class:`pyflink.table.window.GroupWindow`.
"""
def __init__(self, java_group_windowed_table):
@@ -597,6 +639,7 @@ class GroupWindowedTable(object):
Example:
::
+
>>> tab.window(groupWindow.alias("w")).group_by("w, key").select("key, value.avg")
:param fields: Group keys.
@@ -607,7 +650,7 @@ class GroupWindowedTable(object):
class WindowGroupedTable(object):
"""
- A table that has been windowed and grouped for :class:`GroupWindow`s.
+ A table that has been windowed and grouped for :class:`pyflink.table.window.GroupWindow`.
"""
def __init__(self, java_window_grouped_table):
@@ -621,17 +664,18 @@ class WindowGroupedTable(object):
Example:
::
+
>>> window_grouped_table.select("key, window.start, value.avg as valavg")
:param fields: Expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.select(fields))
class OverWindowedTable(object):
"""
- A table that has been windowed for :class:`OverWindow`s.
+ A table that has been windowed for :class:`pyflink.table.window.OverWindow`.
Unlike group windows, which are specified in the GROUP BY clause, over windows do not collapse
rows. Instead over window aggregates compute an aggregate for each input row over a range of
@@ -649,9 +693,10 @@ class OverWindowedTable(object):
Example:
::
+
>>> over_windowed_table.select("c, b.count over ow, e.sum over ow")
:param fields: Expression string.
- :return: Result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_table.select(fields))
diff --git a/flink-python/pyflink/table/table_environment.py b/flink-python/pyflink/table/table_environment.py
index 72f8675..4cb4bf6 100644
--- a/flink-python/pyflink/table/table_environment.py
+++ b/flink-python/pyflink/table/table_environment.py
@@ -23,8 +23,8 @@ from pyflink.serializers import BatchedSerializer, PickleSerializer
from pyflink.table.catalog import Catalog
from pyflink.table.query_config import StreamQueryConfig, BatchQueryConfig, QueryConfig
from pyflink.table.table_config import TableConfig
-from pyflink.table.table_descriptor import (StreamTableDescriptor, ConnectorDescriptor,
- BatchTableDescriptor)
+from pyflink.table.descriptors import (StreamTableDescriptor, ConnectorDescriptor,
+ BatchTableDescriptor)
from pyflink.java_gateway import get_gateway
from pyflink.table import Table
@@ -54,28 +54,35 @@ class TableEnvironment(object):
"""
Creates a table from a table source.
+ Example:
+ ::
+
+ >>> csv_table_source = CsvTableSource(
+ ... csv_file_path, ['a', 'b'], [DataTypes.STRING(), DataTypes.BIGINT()])
+ ... table_env.from_table_source(csv_table_source)
+
:param table_source: The table source used as table.
- :return: The result table.
+ :return: The result :class:`Table`.
"""
return Table(self._j_tenv.fromTableSource(table_source._j_table_source))
def register_catalog(self, catalog_name, catalog):
"""
- Registers a :class:`Catalog` under a unique name.
- All tables registered in the :class:`Catalog` can be accessed.
+ Registers a :class:`pyflink.table.catalog.Catalog` under a unique name.
+ All tables registered in the :class:`pyflink.table.catalog.Catalog` can be accessed.
:param catalog_name: The name under which the catalog will be registered.
- :param catalog: The catalog :class:`Catalog` to register.
+ :param catalog: The :class:`pyflink.table.catalog.Catalog` to register.
"""
self._j_tenv.registerCatalog(catalog_name, catalog._j_catalog)
def get_catalog(self, catalog_name):
"""
- Gets a registered :class:`Catalog` by name.
+ Gets a registered :class:`pyflink.table.catalog.Catalog` by name.
- :param catalog_name: The name to look up the :class:`Catalog`.
- :return: The requested catalog :class:`Catalog`, None if there is no registered catalog
- with given name.
+ :param catalog_name: The name to look up the :class:`pyflink.table.catalog.Catalog`.
+ :return: The requested :class:`pyflink.table.catalog.Catalog`, None if there is no
+ registered catalog with given name.
"""
catalog = self._j_tenv.getCatalog(catalog_name)
if catalog.isPresent():
@@ -125,7 +132,7 @@ class TableEnvironment(object):
"""
Scans a registered table and returns the resulting :class:`Table`.
A table to scan must be registered in the TableEnvironment. It can be either directly
- registered or be an external member of a :class:`Catalog`.
+ registered or be an external member of a :class:`pyflink.table.catalog.Catalog`.
See the documentation of :func:`~pyflink.table.TableEnvironment.use_database` or
:func:`~pyflink.table.TableEnvironment.use_catalog` for the rules on the path resolution.
@@ -134,10 +141,12 @@ class TableEnvironment(object):
Scanning a directly registered table
::
+
>>> tab = t_env.scan("tableName")
Scanning a table from a registered catalog
::
+
>>> tab = t_env.scan("catalogName", "dbName", "tableName")
:param table_path: The path of the table to scan.
@@ -153,14 +162,15 @@ class TableEnvironment(object):
"""
Writes the :class:`Table` to a :class:`TableSink` that was registered under
the specified name. For the path resolution algorithm see
- :func:`~TableEnvironment.useDatabase`.
+ :func:`~TableEnvironment.use_database`.
Example:
::
+
>>> tab = t_env.scan("tableName")
>>> t_env.insert_into(tab, "print")
- :param table The :class:`Table` to write to the sink.
+ :param table: :class:`Table` to write to the sink.
:param table_path: The first part of the path of the registered :class:`TableSink` to which
the :class:`Table` is written. This is to ensure at least the name of the
:class:`Table` is provided.
@@ -200,8 +210,8 @@ class TableEnvironment(object):
called, for example when it is embedded into a String.
Hence, SQL queries can directly reference a :class:`Table` as follows:
-
::
+
>>> table = ...
# the table is not registered to the table environment
>>> t_env.sql_query("SELECT * FROM %s" % table)
@@ -216,15 +226,16 @@ class TableEnvironment(object):
"""
Evaluates a SQL statement such as INSERT, UPDATE or DELETE or a DDL statement
- ..note::
+ .. note::
+
Currently only SQL INSERT statements are supported.
All tables referenced by the query must be registered in the TableEnvironment.
A :class:`Table` is automatically registered when its :func:`~Table.__str__` method is
called, for example when it is embedded into a String.
Hence, SQL queries can directly reference a :class:`Table` as follows:
-
::
+
# register the table sink into which the result is inserted.
>>> t_env.register_table_sink("sink_table", field_names, fields_types, table_sink)
>>> source_table = ...
@@ -244,10 +255,11 @@ class TableEnvironment(object):
"""
Gets the current default catalog name of the current session.
- :return The current default catalog name that is used for the path resolution.
+ :return: The current default catalog name that is used for the path resolution.
+
.. seealso:: :func:`~pyflink.table.TableEnvironment.use_catalog`
"""
- self._j_tenv.getCurrentCatalog()
+ return self._j_tenv.getCurrentCatalog()
def use_catalog(self, catalog_name):
"""
@@ -291,8 +303,9 @@ class TableEnvironment(object):
+----------------+-----------------------------------------+
:param: catalog_name: The name of the catalog to set as the current default catalog.
- :throws: CatalogException thrown if a catalog with given name could not be set as the
- default one
+ :throws: :class:`pyflink.util.exceptions.CatalogException` thrown if a catalog with given
+ name could not be set as the default one.
+
.. seealso:: :func:`~pyflink.table.TableEnvironment.use_database`
"""
self._j_tenv.useCatalog(catalog_name)
@@ -301,10 +314,11 @@ class TableEnvironment(object):
"""
Gets the current default database name of the running session.
- :return The name of the current database of the current catalog.
+ :return: The name of the current database of the current catalog.
+
.. seealso:: :func:`~pyflink.table.TableEnvironment.use_database`
"""
- self._j_tenv.getCurrentCatalog()
+ return self._j_tenv.getCurrentDatabase()
def use_database(self, database_name):
"""
@@ -346,8 +360,9 @@ class TableEnvironment(object):
| cat1.db1.tab1 | cat1.db1.tab1 |
+----------------+-----------------------------------------+
- :throws: CatalogException thrown if the given catalog and database could not be set as
- the default ones
+ :throws: :class:`pyflink.util.exceptions.CatalogException` thrown if the given catalog and
+ database could not be set as the default ones.
+
.. seealso:: :func:`~pyflink.table.TableEnvironment.use_catalog`
:param: database_name: The name of the database to set as the current database.
@@ -376,6 +391,12 @@ class TableEnvironment(object):
@abstractmethod
def query_config(self):
+ """
+ Returns a :class:`StreamQueryConfig` that holds parameters to configure the behavior of
+ streaming queries.
+
+ :return: A new :class:`StreamQueryConfig` or :class:`BatchQueryConfig`.
+ """
pass
@abstractmethod
@@ -389,21 +410,25 @@ class TableEnvironment(object):
The following example shows how to read from a connector using a JSON format and
registering a table source as "MyTable":
+
+ Example:
::
- >>> table_env\
+
+ >>> table_env\\
... .connect(ExternalSystemXYZ()
- ... .version("0.11"))\
+ ... .version("0.11"))\\
... .with_format(Json()
- ... .json_schema("{...}")
- ... .fail_on_missing_field(False))\
+ ... .json_schema("{...}")
+ ... .fail_on_missing_field(False))\\
... .with_schema(Schema()
... .field("user-name", "VARCHAR")
... .from_origin_field("u_name")
- ... .field("count", "DECIMAL"))\
+ ... .field("count", "DECIMAL"))\\
... .register_table_source("MyTable")
:param connector_descriptor: Connector descriptor describing the external system.
- :return: A :class:`ConnectTableDescriptor` used to build the table source/sink.
+ :return: A :class:`pyflink.table.descriptors.ConnectTableDescriptor` used to build the
+ table source/sink.
"""
pass
@@ -437,10 +462,15 @@ class TableEnvironment(object):
"""
Creates a table from a collection of elements.
+ Example:
+ ::
+
+ >>> table_env.from_elements([(1, 'Hi'), (2, 'Hello')], ['a', 'b'])
+
:param elements: The elements to create a table from.
:param schema: The schema of the table.
:param verify_schema: Whether to verify the elements against the schema.
- :return: A Table.
+ :return: The result :class:`Table`.
"""
# verifies the elements against the specified schema
@@ -490,7 +520,7 @@ class TableEnvironment(object):
Creates a table from a collection of elements.
:param elements: The elements to create a table from.
- :return: A table.
+ :return: The result :class:`Table`.
"""
# serializes to a file, and we read the file in java
@@ -555,16 +585,17 @@ class StreamTableEnvironment(TableEnvironment):
The following example shows how to read from a connector using a JSON format and
registering a table source as "MyTable":
::
- >>> table_env\
+
+ >>> table_env\\
... .connect(ExternalSystemXYZ()
- ... .version("0.11"))\
+ ... .version("0.11"))\\
... .with_format(Json()
- ... .json_schema("{...}")
- ... .fail_on_missing_field(False))\
+ ... .json_schema("{...}")
+ ... .fail_on_missing_field(False))\\
... .with_schema(Schema()
... .field("user-name", "VARCHAR")
... .from_origin_field("u_name")
- ... .field("count", "DECIMAL"))\
+ ... .field("count", "DECIMAL"))\\
... .register_table_source("MyTable")
:param connector_descriptor: Connector descriptor describing the external system.
@@ -620,16 +651,17 @@ class BatchTableEnvironment(TableEnvironment):
The following example shows how to read from a connector using a JSON format and
registering a table source as "MyTable":
::
- >>> table_env\
+
+ >>> table_env\\
... .connect(ExternalSystemXYZ()
- ... .version("0.11"))\
+ ... .version("0.11"))\\
... .with_format(Json()
- ... .json_schema("{...}")
- ... .fail_on_missing_field(False))\
+ ... .json_schema("{...}")
+ ... .fail_on_missing_field(False))\\
... .with_schema(Schema()
... .field("user-name", "VARCHAR")
... .from_origin_field("u_name")
- ... .field("count", "DECIMAL"))\
+ ... .field("count", "DECIMAL"))\\
... .register_table_source("MyTable")
:param connector_descriptor: Connector descriptor describing the external system.
diff --git a/flink-python/pyflink/table/tests/test_catalog_completeness.py b/flink-python/pyflink/table/tests/test_catalog_completeness.py
index 677833f..40612e5 100644
--- a/flink-python/pyflink/table/tests/test_catalog_completeness.py
+++ b/flink-python/pyflink/table/tests/test_catalog_completeness.py
@@ -19,7 +19,7 @@
import unittest
from pyflink.testing.test_case_utils import PythonAPICompletenessTestCase
-from pyflink.table import Catalog, CatalogDatabase, CatalogBaseTable, CatalogPartition, \
+from pyflink.table.catalog import Catalog, CatalogDatabase, CatalogBaseTable, CatalogPartition, \
CatalogFunction, CatalogColumnStatistics, CatalogPartitionSpec, ObjectPath
diff --git a/flink-python/pyflink/table/tests/test_descriptor.py b/flink-python/pyflink/table/tests/test_descriptor.py
index 1246666..a8b8a03 100644
--- a/flink-python/pyflink/table/tests/test_descriptor.py
+++ b/flink-python/pyflink/table/tests/test_descriptor.py
@@ -17,8 +17,8 @@
################################################################################
import os
-from pyflink.table.table_descriptor import (FileSystem, OldCsv, Rowtime, Schema, Kafka,
- Elasticsearch, Csv, Avro, Json)
+from pyflink.table.descriptors import (FileSystem, OldCsv, Rowtime, Schema, Kafka,
+ Elasticsearch, Csv, Avro, Json)
from pyflink.table.table_schema import TableSchema
from pyflink.table.types import DataTypes
from pyflink.testing.test_case_utils import (PyFlinkTestCase, PyFlinkStreamTableTestCase,
diff --git a/flink-python/pyflink/table/tests/test_window.py b/flink-python/pyflink/table/tests/test_window.py
index 58428e1..d093c8a 100644
--- a/flink-python/pyflink/table/tests/test_window.py
+++ b/flink-python/pyflink/table/tests/test_window.py
@@ -18,9 +18,8 @@
from py4j.protocol import Py4JJavaError
-from pyflink.table.window import Session, Slide, Tumble
-from pyflink.table import Over
-from pyflink.testing.test_case_utils import PyFlinkBatchTableTestCase, PyFlinkStreamTableTestCase
+from pyflink.table.window import Session, Slide, Tumble, Over
+from pyflink.testing.test_case_utils import PyFlinkStreamTableTestCase, PyFlinkBatchTableTestCase
class StreamTableWindowTests(PyFlinkStreamTableTestCase):
diff --git a/flink-python/pyflink/table/types.py b/flink-python/pyflink/table/types.py
index 088722f..25050c2 100644
--- a/flink-python/pyflink/table/types.py
+++ b/flink-python/pyflink/table/types.py
@@ -1469,30 +1469,34 @@ class Row(tuple):
a named argument to represent the value is None or missing. This should be
explicitly set to None in this case.
- >>> row = Row(name="Alice", age=11)
- >>> row
- Row(age=11, name='Alice')
- >>> row['name'], row['age']
- ('Alice', 11)
- >>> row.name, row.age
- ('Alice', 11)
- >>> 'name' in row
- True
- >>> 'wrong_key' in row
- False
+ ::
+
+ >>> row = Row(name="Alice", age=11)
+ >>> row
+ Row(age=11, name='Alice')
+ >>> row['name'], row['age']
+ ('Alice', 11)
+ >>> row.name, row.age
+ ('Alice', 11)
+ >>> 'name' in row
+ True
+ >>> 'wrong_key' in row
+ False
Row can also be used to create another Row like class, then it
could be used to create Row objects, such as
- >>> Person = Row("name", "age")
- >>> Person
- <Row(name, age)>
- >>> 'name' in Person
- True
- >>> 'wrong_key' in Person
- False
- >>> Person("Alice", 11)
- Row(name='Alice', age=11)
+ ::
+
+ >>> Person = Row("name", "age")
+ >>> Person
+ <Row(name, age)>
+ >>> 'name' in Person
+ True
+ >>> 'wrong_key' in Person
+ False
+ >>> Person("Alice", 11)
+ Row(name='Alice', age=11)
"""
def __new__(cls, *args, **kwargs):
@@ -1515,15 +1519,18 @@ class Row(tuple):
"""
Returns as a dict.
- :param recursive: turns the nested Row as dict (default: False).
+ Example:
+ ::
- >>> Row(name="Alice", age=11).as_dict() == {'name': 'Alice', 'age': 11}
- True
- >>> row = Row(key=1, value=Row(name='a', age=2))
- >>> row.as_dict() == {'key': 1, 'value': Row(age=2, name='a')}
- True
- >>> row.as_dict(True) == {'key': 1, 'value': {'name': 'a', 'age': 2}}
- True
+ >>> Row(name="Alice", age=11).as_dict() == {'name': 'Alice', 'age': 11}
+ True
+ >>> row = Row(key=1, value=Row(name='a', age=2))
+ >>> row.as_dict() == {'key': 1, 'value': Row(age=2, name='a')}
+ True
+ >>> row.as_dict(True) == {'key': 1, 'value': {'name': 'a', 'age': 2}}
+ True
+
+ :param recursive: turns the nested Row as dict (default: False).
"""
if not hasattr(self, "_fields"):
raise TypeError("Cannot convert a Row class into dict")
diff --git a/flink-python/pyflink/table/window.py b/flink-python/pyflink/table/window.py
index 96e495e..218d640 100644
--- a/flink-python/pyflink/table/window.py
+++ b/flink-python/pyflink/table/window.py
@@ -54,6 +54,7 @@ class Tumble(object):
Example:
::
+
>>> Tumble.over("10.minutes").on("rowtime").alias("w")
"""
@@ -131,6 +132,7 @@ class Session(object):
Example:
::
+
>>> Session.with_gap("10.minutes").on("rowtime").alias("w")
"""
@@ -214,6 +216,7 @@ class Slide(object):
Example:
::
+
>>> Slide.over("10.minutes").every("5.minutes").on("rowtime").alias("w")
"""
@@ -321,6 +324,7 @@ class Over(object):
Example:
::
+
>>> Over.partition_by("a").order_by("rowtime").preceding("unbounded_range").alias("w")
"""
diff --git a/flink-python/pyflink/testing/test_case_utils.py b/flink-python/pyflink/testing/test_case_utils.py
index 35a66d7..6e76508 100644
--- a/flink-python/pyflink/testing/test_case_utils.py
+++ b/flink-python/pyflink/testing/test_case_utils.py
@@ -25,7 +25,7 @@ import unittest
from abc import abstractmethod
from py4j.java_gateway import JavaObject
-from pyflink.table.table_source import CsvTableSource
+from pyflink.table.sources import CsvTableSource
from pyflink.find_flink_home import _find_flink_home
from pyflink.table import TableEnvironment, TableConfig