Google Git Google Git
Sign in
apache/flink/refs/heads/release-2.4/./flink-python/pyflink/table/table_environment.py
blob: ebab2034a38536704a39126cafcf0db2523a7041 [file]
################################################################################
# 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.
################################################################################
import atexit
import os
import sys
import tempfile
from typing import Union, List, Tuple, Iterable, Optional, TYPE_CHECKING
if TYPE_CHECKING:
import pandas
from py4j.java_gateway import get_java_class, get_method
from pyflink.common.configuration import Configuration
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common.typeinfo import TypeInformation
from pyflink.datastream.data_stream import DataStream
from pyflink.java_gateway import get_gateway
from pyflink.serializers import BatchedSerializer, PickleSerializer
from pyflink.table import Table, EnvironmentSettings, Expression, ExplainDetail, \
Module, ModuleEntry, Schema, ChangelogMode
from pyflink.table.catalog import Catalog, CatalogDescriptor
from pyflink.table.model import Model
from pyflink.table.model_descriptor import ModelDescriptor
from pyflink.table.compiled_plan import CompiledPlan
from pyflink.table.plan_reference import PlanReference
from pyflink.table.serializers import ArrowSerializer
from pyflink.table.statement_set import StatementSet
from pyflink.table.table_config import TableConfig
from pyflink.table.table_descriptor import TableDescriptor
from pyflink.table.table_result import TableResult
from pyflink.table.types import _create_type_verifier, RowType, DataType, TimestampType, \
_infer_schema_from_data, _create_converter, from_arrow_type, RowField, create_arrow_schema, \
to_arrow_type, _to_java_data_type
from pyflink.table.udf import UserDefinedFunctionWrapper, AggregateFunction, udaf, \
udtaf, TableAggregateFunction
from pyflink.table.utils import to_expression_jarray
from pyflink.util.api_stability_decorators import PublicEvolving, Deprecated
from pyflink.util.java_utils import get_j_env_configuration, is_local_deployment, load_java_class, \
to_j_explain_detail_arr, to_jarray, get_field
__all__ = [
'StreamTableEnvironment',
'TableEnvironment'
]
_TIMESTAMP_UNITS_PER_SECOND = {
's': 1,
'ms': 1_000,
'us': 1_000_000,
'ns': 1_000_000_000,
}
def _cast_arrow_timestamp(array, target_type):
import pyarrow as pa
import pyarrow.compute as pc
if isinstance(array, pa.ChunkedArray):
chunks = [
_cast_arrow_timestamp(chunk, target_type) for chunk in array.chunks
]
return pa.chunked_array(chunks, type=target_type)
if array.type == target_type:
return array
source_units = _TIMESTAMP_UNITS_PER_SECOND[array.type.unit]
target_units = _TIMESTAMP_UNITS_PER_SECOND[target_type.unit]
values = array.view(pa.int64())
if source_units > target_units:
divisor = source_units // target_units
quotient = pc.divide_checked(values, divisor)
remainder = pc.subtract_checked(
values, pc.multiply_checked(quotient, divisor))
# Arrow integer division truncates toward zero. Timestamp precision reduction
# needs floor division to preserve the local date and time before the epoch.
values = pc.subtract_checked(
quotient,
pc.cast(pc.less(remainder, 0), pa.int64()),
)
elif source_units < target_units:
values = pc.multiply_checked(values, target_units // source_units)
return values.view(target_type)
def _contains_timezone_aware_timestamp(arrow_type):
import pyarrow as pa
if pa.types.is_timestamp(arrow_type):
return arrow_type.tz is not None
if pa.types.is_struct(arrow_type):
return any(
_contains_timezone_aware_timestamp(field.type) for field in arrow_type
)
if pa.types.is_list(arrow_type):
return _contains_timezone_aware_timestamp(arrow_type.value_type)
if pa.types.is_map(arrow_type):
return (
_contains_timezone_aware_timestamp(arrow_type.key_type)
or _contains_timezone_aware_timestamp(arrow_type.item_type)
)
return False
@PublicEvolving()
class TableEnvironment(object):
"""
A table environment is the base class, entry point, and central context for creating Table
and SQL API programs.
It is unified for bounded and unbounded data processing.
A table environment is responsible for:
- Connecting to external systems.
- Registering and retrieving :class:`~pyflink.table.Table` and other meta objects from a
catalog.
- Executing SQL statements.
- Offering further configuration options.
The path in methods such as :func:`create_temporary_view`
should be a proper SQL identifier. The syntax is following
[[catalog-name.]database-name.]object-name, where the catalog name and database are optional.
For path resolution see :func:`use_catalog` and :func:`use_database`. All keywords or other
special characters need to be escaped.
Example: `cat.1`.`db`.`Table` resolves to an object named 'Table' (table is a reserved
keyword, thus must be escaped) in a catalog named 'cat.1' and database named 'db'.
.. note::
This environment is meant for pure table programs. If you would like to convert from or to
other Flink APIs, it might be necessary to use one of the available language-specific table
environments in the corresponding bridging modules.
"""
def __init__(self, j_tenv, serializer=PickleSerializer()):
self._j_tenv = j_tenv
self._serializer = serializer
# When running in MiniCluster, launch the Python UDF worker using the Python executable
# specified by sys.executable if users have not specified it explicitly via configuration
# python.executable.
self._set_python_executable_for_local_executor()
self._config_chaining_optimization()
self._open()
@staticmethod
def create(environment_settings: Union[EnvironmentSettings, Configuration]) \
-> 'TableEnvironment':
"""
Creates a table environment that is the entry point and central context for creating Table
and SQL API programs.
:param environment_settings: The configuration or environment settings used to instantiate
the :class:`~pyflink.table.TableEnvironment`, the name is for backward compatibility.
:return: The :class:`~pyflink.table.TableEnvironment`.
"""
gateway = get_gateway()
if isinstance(environment_settings, Configuration):
environment_settings = EnvironmentSettings.new_instance() \
.with_configuration(environment_settings).build()
elif not isinstance(environment_settings, EnvironmentSettings):
raise TypeError("argument should be EnvironmentSettings or Configuration")
j_tenv = gateway.jvm.TableEnvironment.create(environment_settings._j_environment_settings)
return TableEnvironment(j_tenv)
def create_catalog(self, catalog_name: str, catalog_descriptor: CatalogDescriptor):
"""
Creates a :class:`~pyflink.table.catalog.Catalog` using the provided
:class:`~pyflink.table.catalog.CatalogDescriptor`. All table registered in the
:class:`~pyflink.table.catalog.Catalog` can be accessed.
:param catalog_name: The name under which the catalog will be created.
:param catalog_descriptor: The catalog descriptor for creating catalog.
"""
self._j_tenv.createCatalog(catalog_name, catalog_descriptor._j_catalog_descriptor)
@Deprecated(since="2.1.0", detail="""
Use :func:`create_catalog` instead. The new method uses a
:class:`~pyflink.table.catalog.CatalogDescriptor` to initialize the catalog instance and store
the :class:`~pyflink.table.catalog.CatalogDescriptor` in the catalog store.
""")
def register_catalog(self, catalog_name: str, catalog: Catalog):
"""
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 to register.
"""
self._j_tenv.registerCatalog(catalog_name, catalog._j_catalog)
def get_catalog(self, catalog_name: str) -> Catalog:
"""
Gets a registered :class:`~pyflink.table.catalog.Catalog` by name.
:param catalog_name: The name to look up the :class:`~pyflink.table.catalog.Catalog`.
:return: The requested catalog, None if there is no
registered catalog with given name.
"""
catalog = self._j_tenv.getCatalog(catalog_name)
if catalog.isPresent():
return Catalog(catalog.get())
else:
return None
def load_module(self, module_name: str, module: Module):
"""
Loads a :class:`~pyflink.table.Module` under a unique name. Modules will be kept
in the loaded order.
ValidationException is thrown when there is already a module with the same name.
:param module_name: Name of the :class:`~pyflink.table.Module`.
:param module: The module instance.
.. versionadded:: 1.12.0
"""
self._j_tenv.loadModule(module_name, module._j_module)
def unload_module(self, module_name: str):
"""
Unloads a :class:`~pyflink.table.Module` with given name.
ValidationException is thrown when there is no module with the given name.
:param module_name: Name of the :class:`~pyflink.table.Module`.
.. versionadded:: 1.12.0
"""
self._j_tenv.unloadModule(module_name)
def use_modules(self, *module_names: str):
"""
Use an array of :class:`~pyflink.table.Module` with given names.
ValidationException is thrown when there is duplicate name or no module with the given name.
:param module_names: Names of the modules to be used.
.. versionadded:: 1.13.0
"""
j_module_names = to_jarray(get_gateway().jvm.String, module_names)
self._j_tenv.useModules(j_module_names)
def create_java_temporary_system_function(self, name: str, function_class_name: str):
"""
Registers a java user defined function class as a temporary system function.
Compared to .. seealso:: :func:`create_java_temporary_function`, system functions are
identified by a global name that is independent of the current catalog and current
database. Thus, this method allows to extend the set of built-in system functions like
TRIM, ABS, etc.
Temporary functions can shadow permanent ones. If a permanent function under a given name
exists, it will be inaccessible in the current session. To make the permanent function
available again one can drop the corresponding temporary system function.
Example:
::
>>> table_env.create_java_temporary_system_function("func",
... "java.user.defined.function.class.name")
:param name: The name under which the function will be registered globally.
:param function_class_name: The java full qualified class name of the function class
containing the implementation. The function must have a
public no-argument constructor and can be founded in current
Java classloader.
.. versionadded:: 1.12.0
"""
gateway = get_gateway()
java_function = gateway.jvm.Thread.currentThread().getContextClassLoader() \
.loadClass(function_class_name)
self._j_tenv.createTemporarySystemFunction(name, java_function)
def create_temporary_system_function(self, name: str,
function: Union[UserDefinedFunctionWrapper,
AggregateFunction]):
"""
Registers a python user defined function class as a temporary system function.
Compared to .. seealso:: :func:`create_temporary_function`, system functions are identified
by a global name that is independent of the current catalog and current database. Thus,
this method allows to extend the set of built-in system functions like TRIM, ABS, etc.
Temporary functions can shadow permanent ones. If a permanent function under a given name
exists, it will be inaccessible in the current session. To make the permanent function
available again one can drop the corresponding temporary system function.
Example:
::
>>> table_env.create_temporary_system_function(
... "add_one", udf(lambda i: i + 1, result_type=DataTypes.BIGINT()))
>>> @udf(result_type=DataTypes.BIGINT())
... def add(i, j):
... return i + j
>>> table_env.create_temporary_system_function("add", add)
>>> class SubtractOne(ScalarFunction):
... def eval(self, i):
... return i - 1
>>> table_env.create_temporary_system_function(
... "subtract_one", udf(SubtractOne(), result_type=DataTypes.BIGINT()))
:param name: The name under which the function will be registered globally.
:param function: The function class containing the implementation. The function must have a
public no-argument constructor and can be founded in current Java
classloader.
.. versionadded:: 1.12.0
"""
function = self._wrap_aggregate_function_if_needed(function)
java_function = function._java_user_defined_function()
self._j_tenv.createTemporarySystemFunction(name, java_function)
def drop_temporary_system_function(self, name: str) -> bool:
"""
Drops a temporary system function registered under the given name.
If a permanent function with the given name exists, it will be used from now on for any
queries that reference this name.
:param name: The name under which the function has been registered globally.
:return: true if a function existed under the given name and was removed.
.. versionadded:: 1.12.0
"""
return self._j_tenv.dropTemporarySystemFunction(name)
def create_java_function(self, path: str, function_class_name: str,
ignore_if_exists: bool = None):
"""
Registers a java user defined function class as a catalog function in the given path.
Compared to system functions with a globally defined name, catalog functions are always
(implicitly or explicitly) identified by a catalog and database.
There must not be another function (temporary or permanent) registered under the same path.
Example:
::
>>> table_env.create_java_function("func", "java.user.defined.function.class.name")
:param path: The path under which the function will be registered.
See also the :class:`~pyflink.table.TableEnvironment` class description for
the format of the path.
:param function_class_name: The java full qualified class name of the function class
containing the implementation. The function must have a
public no-argument constructor and can be founded in current
Java classloader.
:param ignore_if_exists: If a function exists under the given path and this flag is set,
no operation is executed. An exception is thrown otherwise.
.. versionadded:: 1.12.0
"""
gateway = get_gateway()
java_function = gateway.jvm.Thread.currentThread().getContextClassLoader() \
.loadClass(function_class_name)
if ignore_if_exists is None:
self._j_tenv.createFunction(path, java_function)
else:
self._j_tenv.createFunction(path, java_function, ignore_if_exists)
def drop_function(self, path: str) -> bool:
"""
Drops a catalog function registered in the given path.
:param path: The path under which the function will be registered.
See also the :class:`~pyflink.table.TableEnvironment` class description for
the format of the path.
:return: true if a function existed in the given path and was removed.
.. versionadded:: 1.12.0
"""
return self._j_tenv.dropFunction(path)
def create_java_temporary_function(self, path: str, function_class_name: str):
"""
Registers a java user defined function class as a temporary catalog function.
Compared to .. seealso:: :func:`create_java_temporary_system_function` with a globally
defined name, catalog functions are always (implicitly or explicitly) identified by a
catalog and database.
Temporary functions can shadow permanent ones. If a permanent function under a given name
exists, it will be inaccessible in the current session. To make the permanent function
available again one can drop the corresponding temporary function.
Example:
::
>>> table_env.create_java_temporary_function("func",
... "java.user.defined.function.class.name")
:param path: The path under which the function will be registered.
See also the :class:`~pyflink.table.TableEnvironment` class description for
the format of the path.
:param function_class_name: The java full qualified class name of the function class
containing the implementation. The function must have a
public no-argument constructor and can be founded in current
Java classloader.
.. versionadded:: 1.12.0
"""
gateway = get_gateway()
java_function = gateway.jvm.Thread.currentThread().getContextClassLoader() \
.loadClass(function_class_name)
self._j_tenv.createTemporaryFunction(path, java_function)
def create_temporary_function(self, path: str, function: Union[UserDefinedFunctionWrapper,
AggregateFunction]):
"""
Registers a python user defined function class as a temporary catalog function.
Compared to .. seealso:: :func:`create_temporary_system_function` with a globally defined
name, catalog functions are always (implicitly or explicitly) identified by a catalog and
database.
Temporary functions can shadow permanent ones. If a permanent function under a given name
exists, it will be inaccessible in the current session. To make the permanent function
available again one can drop the corresponding temporary function.
Example:
::
>>> table_env.create_temporary_function(
... "add_one", udf(lambda i: i + 1, result_type=DataTypes.BIGINT()))
>>> @udf(result_type=DataTypes.BIGINT())
... def add(i, j):
... return i + j
>>> table_env.create_temporary_function("add", add)
>>> class SubtractOne(ScalarFunction):
... def eval(self, i):
... return i - 1
>>> table_env.create_temporary_function(
... "subtract_one", udf(SubtractOne(), result_type=DataTypes.BIGINT()))
:param path: The path under which the function will be registered.
See also the :class:`~pyflink.table.TableEnvironment` class description for
the format of the path.
:param function: The function class containing the implementation. The function must have a
public no-argument constructor and can be founded in current Java
classloader.
.. versionadded:: 1.12.0
"""
function = self._wrap_aggregate_function_if_needed(function)
java_function = function._java_user_defined_function()
self._j_tenv.createTemporaryFunction(path, java_function)
def drop_temporary_function(self, path: str) -> bool:
"""
Drops a temporary system function registered under the given name.
If a permanent function with the given name exists, it will be used from now on for any
queries that reference this name.
:param path: The path under which the function will be registered.
See also the :class:`~pyflink.table.TableEnvironment` class description for
the format of the path.
:return: true if a function existed in the given path and was removed.
.. versionadded:: 1.12.0
"""
return self._j_tenv.dropTemporaryFunction(path)
def create_temporary_table(self, path: str, descriptor: TableDescriptor,
ignoreIfExists: Optional[bool] = False):
"""
Registers the given :class:`~pyflink.table.TableDescriptor` as a temporary catalog table.
The TableDescriptor is converted into a CatalogTable and stored in the catalog.
Temporary objects can shadow permanent ones. If a permanent object in a given path exists,
it will be inaccessible in the current session. To make the permanent object available again
one can drop the corresponding temporary object.
Examples:
::
>>> table_env.create_temporary_table("MyTable", TableDescriptor.for_connector("datagen")
... .schema(Schema.new_builder()
... .column("f0", DataTypes.STRING())
... .build())
... .option("rows-per-second", 10)
... .option("fields.f0.kind", "random")
... .build(),
... True)
:param path: The path under which the table will be registered.
:param descriptor: Template for creating a CatalogTable instance.
:param ignoreIfExists: If a table exists under the given path and this flag is set,
no operation is executed. An exception is thrown otherwise.
.. versionadded:: 1.14.0
"""
self._j_tenv.createTemporaryTable(path, descriptor._j_table_descriptor, ignoreIfExists)
def create_table(self, path: str, descriptor: TableDescriptor,
ignoreIfExists: Optional[bool] = False):
"""
Registers the given :class:`~pyflink.table.TableDescriptor` as a catalog table.
The TableDescriptor is converted into a CatalogTable and stored in the catalog.
If the table should not be permanently stored in a catalog, use
:func:`create_temporary_table` instead.
Examples:
::
>>> table_env.create_table("MyTable", TableDescriptor.for_connector("datagen")
... .schema(Schema.new_builder()
... .column("f0", DataTypes.STRING())
... .build())
... .option("rows-per-second", 10)
... .option("fields.f0.kind", "random")
... .build(),
... True)
:param path: The path under which the table will be registered.
:param descriptor: Template for creating a CatalogTable instance.
:param ignoreIfExists: If a table exists under the given path and this flag is set,
no operation is executed. An exception is thrown otherwise.
.. versionadded:: 1.14.0
"""
self._j_tenv.createTable(path, descriptor._j_table_descriptor, ignoreIfExists)
def from_path(self, path: str) -> Table:
"""
Reads a registered table and returns the resulting :class:`~pyflink.table.Table`.
A table to scan must be registered in the :class:`~pyflink.table.TableEnvironment`.
See the documentation of :func:`use_database` or :func:`use_catalog` for the rules on the
path resolution.
Examples:
Reading a table from default catalog and database.
::
>>> tab = table_env.from_path("tableName")
Reading a table from a registered catalog.
::
>>> tab = table_env.from_path("catalogName.dbName.tableName")
Reading a table from a registered catalog with escaping. (`Table` is a reserved keyword).
Dots in e.g. a database name also must be escaped.
::
>>> tab = table_env.from_path("catalogName.`db.Name`.`Table`")
:param path: The path of a table API object to scan.
:return: Either a table or virtual table (=view).
.. seealso:: :func:`use_catalog`
.. seealso:: :func:`use_database`
.. versionadded:: 1.10.0
"""
return Table(get_method(self._j_tenv, "from")(path), self)
def from_descriptor(self, descriptor: TableDescriptor) -> Table:
"""
Returns a Table backed by the given TableDescriptor.
The TableDescriptor is registered as an inline (i.e. anonymous) temporary table
(see :func:`create_temporary_table`) using a unique identifier and then read. Note that
calling this method multiple times, even with the same descriptor, results in multiple
temporary tables. In such cases, it is recommended to register it under a name using
:func:`create_temporary_table` and reference it via :func:`from_path`
Examples:
::
>>> table_env.from_descriptor(TableDescriptor.for_connector("datagen")
... .schema(Schema.new_builder()
... .column("f0", DataTypes.STRING())
... .build())
... .build()
Note that the returned Table is an API object and only contains a pipeline description.
It actually corresponds to a <i>view</i> in SQL terms. Call :func:`execute` in Table to
trigger an execution.
:return: The Table object describing the pipeline for further transformations.
.. versionadded:: 1.14.0
"""
return Table(get_method(self._j_tenv, "from")(descriptor._j_table_descriptor), self)
def list_catalogs(self) -> List[str]:
"""
Gets the names of all catalogs registered in this environment.
:return: List of catalog names.
"""
j_catalog_name_array = self._j_tenv.listCatalogs()
return [item for item in j_catalog_name_array]
def list_modules(self) -> List[str]:
"""
Gets the names of all modules used in this environment.
:return: List of module names.
.. versionadded:: 1.10.0
"""
j_module_name_array = self._j_tenv.listModules()
return [item for item in j_module_name_array]
def list_full_modules(self) -> List[ModuleEntry]:
"""
Gets the names and statuses of all modules loaded in this environment.
:return: List of module names and use statuses.
.. versionadded:: 1.13.0
"""
j_module_entry_array = self._j_tenv.listFullModules()
return [ModuleEntry(entry.name(), entry.used()) for entry in j_module_entry_array]
def list_databases(self) -> List[str]:
"""
Gets the names of all databases in the current catalog.
:return: List of database names in the current catalog.
"""
j_database_name_array = self._j_tenv.listDatabases()
return [item for item in j_database_name_array]
def list_tables(self) -> List[str]:
"""
Gets the names of all tables and views in the current database of the current catalog.
It returns both temporary and permanent tables and views.
:return: List of table and view names in the current database of the current catalog.
"""
j_table_name_array = self._j_tenv.listTables()
return [item for item in j_table_name_array]
def list_views(self) -> List[str]:
"""
Gets the names of all views in the current database of the current catalog.
It returns both temporary and permanent views.
:return: List of view names in the current database of the current catalog.
.. versionadded:: 1.11.0
"""
j_view_name_array = self._j_tenv.listViews()
return [item for item in j_view_name_array]
def list_materialized_tables(self) -> List[str]:
"""
Gets the names of all materialized tables available
in the current namespace (the current database of the current catalog).
:return: A list of the names of all registered materialized tables
in the current database of the current catalog.
.. versionadded:: 2.2.0
"""
j_materialized_table_name_array = self._j_tenv.listMaterializedTables()
return [item for item in j_materialized_table_name_array]
def list_user_defined_functions(self) -> List[str]:
"""
Gets the names of all user defined functions registered in this environment.
:return: List of the names of all user defined functions registered in this environment.
"""
j_udf_name_array = self._j_tenv.listUserDefinedFunctions()
return [item for item in j_udf_name_array]
def list_functions(self) -> List[str]:
"""
Gets the names of all functions in this environment.
:return: List of the names of all functions in this environment.
.. versionadded:: 1.10.0
"""
j_function_name_array = self._j_tenv.listFunctions()
return [item for item in j_function_name_array]
def list_temporary_tables(self) -> List[str]:
"""
Gets the names of all temporary tables and views available in the current namespace
(the current database of the current catalog).
:return: A list of the names of all registered temporary tables and views in the current
database of the current catalog.
.. seealso:: :func:`list_tables`
.. versionadded:: 1.10.0
"""
j_table_name_array = self._j_tenv.listTemporaryTables()
return [item for item in j_table_name_array]
def list_temporary_views(self) -> List[str]:
"""
Gets the names of all temporary views available in the current namespace (the current
database of the current catalog).
:return: A list of the names of all registered temporary views in the current database
of the current catalog.
.. seealso:: :func:`list_tables`
.. versionadded:: 1.10.0
"""
j_view_name_array = self._j_tenv.listTemporaryViews()
return [item for item in j_view_name_array]
def list_models(self) -> List[str]:
"""
Gets the names of all model available in the current namespace (the current
database of the current catalog).
:return: A list of the names of all registered models in the current database
of the current catalog.
.. versionadded:: 2.1.0
"""
j_model_name_array = self._j_tenv.listModels()
return [item for item in j_model_name_array]
def list_temporary_models(self) -> List[str]:
"""
Gets the names of all temporary models available in the current namespace (the current
database of the current catalog).
:return: A list of the names of all registered temporary models in the current database
of the current catalog.
.. versionadded:: 2.1.0
"""
j_model_name_array = self._j_tenv.listTemporaryModels()
return [item for item in j_model_name_array]
def drop_temporary_table(self, table_path: str) -> bool:
"""
Drops a temporary table registered in the given path.
If a permanent table with a given path exists, it will be used
from now on for any queries that reference this path.
:param table_path: The path of the registered temporary table.
:return: True if a table existed in the given path and was removed.
.. versionadded:: 1.10.0
"""
return self._j_tenv.dropTemporaryTable(table_path)
def drop_table(self, table_path: str, ignore_if_not_exists: Optional[bool] = True) -> bool:
"""
Drops a table registered in the given path.
This method can only drop permanent objects. Temporary objects can shadow permanent ones.
If a temporary object exists in a given path,
make sure to drop the temporary object first using :func:`drop_temporary_table`.
:param table_path: The path of the registered table.
:param ignore_if_not_exists: Ignore if table does not exist.
:return: True if a table existed in the given path and was removed.
.. versionadded:: 2.0.0
"""
return self._j_tenv.dropTable(table_path, ignore_if_not_exists)
def drop_temporary_view(self, view_path: str) -> bool:
"""
Drops a temporary view registered in the given path.
If a permanent table or view with a given path exists, it will be used
from now on for any queries that reference this path.
:param view_path: The path of the registered temporary view.
:return: True if a view existed in the given path and was removed.
.. versionadded:: 1.10.0
"""
return self._j_tenv.dropTemporaryView(view_path)
def drop_view(self, view_path: str, ignore_if_not_exists: Optional[bool] = True) -> bool:
"""
Drops a view registered in the given path.
This method can only drop permanent objects. Temporary objects can shadow permanent ones.
If a temporary object exists in a given path,
make sure to drop the temporary object first using :func:`drop_temporary_view`.
:param view_path: The path of the registered view.
:param ignore_if_not_exists: Ignore if view does not exist.
:return: True if a view existed in the given path and was removed
.. versionadded:: 2.0.0
"""
return self._j_tenv.dropView(view_path, ignore_if_not_exists)
def drop_temporary_model(self, model_path: str) -> bool:
"""
Drops a temporary model registered in the given path.
If a permanent model with a given path exists, it will be used
from now on for any queries that reference this path.
:param model_path: The path of the registered temporary model.
:return: True if a model existed in the given path and was removed.
.. versionadded:: 2.1.0
"""
return self._j_tenv.dropTemporaryModel(model_path)
def drop_model(self, model_path: str, ignore_if_not_exists: Optional[bool] = True) -> bool:
"""
Drops a model registered in the given path.
This method can only drop permanent objects. Temporary objects can shadow permanent ones.
If a temporary object exists in a given path,
make sure to drop the temporary object first using :func:`drop_temporary_model`.
:param model_path: The path of the registered model.
:param ignore_if_not_exists: Ignore if model does not exist.
:return: True if a model existed in the given path and was removed.
.. versionadded:: 2.1.0
"""
return self._j_tenv.dropModel(model_path, ignore_if_not_exists)
def explain_sql(self, stmt: str, *extra_details: ExplainDetail) -> str:
"""
Returns the AST of the specified statement and the execution plan.
:param stmt: The statement for which the AST and execution plan will be returned.
:param extra_details: The extra explain details which the explain result should include,
e.g. estimated cost, changelog mode for streaming
:return: The statement for which the AST and execution plan will be returned.
.. versionadded:: 1.11.0
"""
JExplainFormat = get_gateway().jvm.org.apache.flink.table.api.ExplainFormat
j_extra_details = to_j_explain_detail_arr(extra_details)
return self._j_tenv.explainSql(stmt, JExplainFormat.TEXT, j_extra_details)
def sql_query(self, query: str) -> Table:
"""
Evaluates a SQL query on registered tables and retrieves the result as a
:class:`~pyflink.table.Table`.
All tables referenced by the query must be registered in the TableEnvironment.
A :class:`~pyflink.table.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:`~pyflink.table.Table` as follows:
::
>>> table = ...
# the table is not registered to the table environment
>>> table_env.sql_query("SELECT * FROM %s" % table)
:param query: The sql query string.
:return: The result table.
"""
j_table = self._j_tenv.sqlQuery(query)
return Table(j_table, self)
def execute_sql(self, stmt: str) -> TableResult:
"""
Execute the given single statement, and return the execution result.
The statement can be DDL/DML/DQL/SHOW/DESCRIBE/EXPLAIN/USE.
For DML and DQL, this method returns TableResult once the job has been submitted.
For DDL and DCL statements, TableResult is returned once the operation has finished.
:return content for DQL/SHOW/DESCRIBE/EXPLAIN,
the affected row count for `DML` (-1 means unknown),
or a string message ("OK") for other statements.
.. versionadded:: 1.11.0
"""
self._before_execute()
return TableResult(self._j_tenv.executeSql(stmt))
def create_statement_set(self) -> StatementSet:
"""
Create a StatementSet instance which accepts DML statements or Tables,
the planner can optimize all added statements and Tables together
and then submit as one job.
:return statement_set instance
.. versionadded:: 1.11.0
"""
_j_statement_set = self._j_tenv.createStatementSet()
return StatementSet(_j_statement_set, self)
def get_current_catalog(self) -> str:
"""
Gets the current default catalog name of the current session.
:return: The current default catalog name that is used for the path resolution.
.. seealso:: :func:`~pyflink.table.TableEnvironment.use_catalog`
"""
return self._j_tenv.getCurrentCatalog()
def use_catalog(self, catalog_name: str):
"""
Sets the current catalog to the given value. It also sets the default
database to the catalog's default one.
See also :func:`~TableEnvironment.use_database`.
This is used during the resolution of object paths. Both the catalog and database are
optional when referencing catalog objects such as tables, views etc. The algorithm looks for
requested objects in following paths in that order:
* ``[current-catalog].[current-database].[requested-path]``
* ``[current-catalog].[requested-path]``
* ``[requested-path]``
Example:
Given structure with default catalog set to ``default_catalog`` and default database set to
``default_database``. ::
root:
|- default_catalog
|- default_database
|- tab1
|- db1
|- tab1
|- cat1
|- db1
|- tab1
The following table describes resolved paths:
+----------------+-----------------------------------------+
| Requested path | Resolved path |
+================+=========================================+
| tab1 | default_catalog.default_database.tab1 |
+----------------+-----------------------------------------+
| db1.tab1 | default_catalog.db1.tab1 |
+----------------+-----------------------------------------+
| cat1.db1.tab1 | cat1.db1.tab1 |
+----------------+-----------------------------------------+
:param catalog_name: The name of the catalog to set as the current default catalog.
: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)
def get_current_database(self) -> str:
"""
Gets the current default database name of the running session.
:return: The name of the current database of the current catalog.
.. seealso:: :func:`~pyflink.table.TableEnvironment.use_database`
"""
return self._j_tenv.getCurrentDatabase()
def use_database(self, database_name: str):
"""
Sets the current default database. It has to exist in the current catalog. That path will
be used as the default one when looking for unqualified object names.
This is used during the resolution of object paths. Both the catalog and database are
optional when referencing catalog objects such as tables, views etc. The algorithm looks for
requested objects in following paths in that order:
* ``[current-catalog].[current-database].[requested-path]``
* ``[current-catalog].[requested-path]``
* ``[requested-path]``
Example:
Given structure with default catalog set to ``default_catalog`` and default database set to
``default_database``. ::
root:
|- default_catalog
|- default_database
|- tab1
|- db1
|- tab1
|- cat1
|- db1
|- tab1
The following table describes resolved paths:
+----------------+-----------------------------------------+
| Requested path | Resolved path |
+================+=========================================+
| tab1 | default_catalog.default_database.tab1 |
+----------------+-----------------------------------------+
| db1.tab1 | default_catalog.db1.tab1 |
+----------------+-----------------------------------------+
| cat1.db1.tab1 | cat1.db1.tab1 |
+----------------+-----------------------------------------+
: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.
"""
self._j_tenv.useDatabase(database_name)
def get_config(self) -> TableConfig:
"""
Returns the table config to define the runtime behavior of the Table API.
:return: Current table config.
"""
if not hasattr(self, "table_config"):
table_config = TableConfig()
table_config._j_table_config = self._j_tenv.getConfig()
setattr(self, "table_config", table_config)
return getattr(self, "table_config")
def create_temporary_view(self,
view_path: str,
table_or_data_stream: Union[Table, DataStream],
*fields_or_schema: Union[str, Expression, Schema]):
"""
1. When table_or_data_stream is a :class:`~pyflink.table.Table`:
Registers a :class:`~pyflink.table.Table` API object as a temporary view similar to SQL
temporary views.
Temporary objects can shadow permanent ones. If a permanent object in a given path
exists, it will be inaccessible in the current session. To make the permanent object
available again you can drop the corresponding temporary object.
2. When table_or_data_stream is a :class:`~pyflink.datastream.DataStream`:
2.1 When fields_or_schema is a str or a sequence of :class:`~pyflink.table.Expression`:
Creates a view from the given {@link DataStream} in a given path with specified
field names. Registered views can be referenced in SQL queries.
1. Reference input fields by name: All fields in the schema definition are
referenced by name (and possibly renamed using an alias (as). Moreover, we can
define proctime and rowtime attributes at arbitrary positions using arbitrary names
(except those that exist in the result schema). In this mode, fields can be
reordered and projected out. This mode can be used for any input type, including
POJOs.
Example:
::
>>> stream = ...
# reorder the fields, rename the original 'f0' field to 'name' and add
# event-time attribute named 'rowtime'
# use str
>>> table_env.create_temporary_view(
... "cat.db.myTable",
... stream,
... "f1, rowtime.rowtime, f0 as 'name'")
# or use a sequence of expression
>>> table_env.create_temporary_view(
... "cat.db.myTable",
... stream,
... col("f1"),
... col("rowtime").rowtime,
... col("f0").alias('name'))
2. Reference input fields by position: In this mode, fields are simply renamed.
Event-time attributes can replace the field on their position in the input data
(if it is of correct type) or be appended at the end. Proctime attributes must be
appended at the end. This mode can only be used if the input type has a defined
field order (tuple, case class, Row) and none of the {@code fields} references a
field of the input type.
Example:
::
>>> stream = ...
# rename the original fields to 'a' and 'b' and extract the internally attached
# timestamp into an event-time attribute named 'rowtime'
# use str
>>> table_env.create_temporary_view(
... "cat.db.myTable", stream, "a, b, rowtime.rowtime")
# or use a sequence of expressions
>>> table_env.create_temporary_view(
... "cat.db.myTable",
... stream,
... col("a"),
... col("b"),
... col("rowtime").rowtime)
Temporary objects can shadow permanent ones. If a permanent object in a given path
exists, it will be inaccessible in the current session. To make the permanent object
available again you can drop the corresponding temporary object.
2.2 When fields_or_schema is a :class:`~pyflink.table.Schema`:
Creates a view from the given {@link DataStream} in a given path. Registered views
can be referenced in SQL queries.
See :func:`from_data_stream` for more information on how a
:class:`~pyflink.datastream.DataStream` is translated into a table.
Temporary objects can shadow permanent ones. If a permanent object in a given path
exists, it will be inaccessible in the current session. To make the permanent object
available again you can drop the corresponding temporary object.
.. note:: create_temporary_view by providing a Schema (case 2.) was added from flink
1.14.0.
:param view_path: The path under which the view will be registered. See also the
:class:`~pyflink.table.TableEnvironment` class description for the format
of the path.
:param table_or_data_stream: The Table or DataStream out of which to create the view.
:param fields_or_schema: The fields expressions(str) to map original fields of the
DataStream to the fields of the View or the customized schema for the final
table.
.. versionadded:: 1.10.0
"""
if isinstance(table_or_data_stream, Table):
self._j_tenv.createTemporaryView(view_path, table_or_data_stream._j_table)
else:
j_data_stream = table_or_data_stream._j_data_stream
JPythonConfigUtil = get_gateway().jvm.org.apache.flink.python.util.PythonConfigUtil
JPythonConfigUtil.configPythonOperator(j_data_stream.getExecutionEnvironment())
if len(fields_or_schema) == 0:
self._j_tenv.createTemporaryView(view_path, j_data_stream)
elif len(fields_or_schema) == 1 and isinstance(fields_or_schema[0], str):
self._j_tenv.createTemporaryView(
view_path,
j_data_stream,
fields_or_schema[0])
elif len(fields_or_schema) == 1 and isinstance(fields_or_schema[0], Schema):
self._j_tenv.createTemporaryView(
view_path,
j_data_stream,
fields_or_schema[0]._j_schema)
elif (len(fields_or_schema) > 0 and
all(isinstance(elem, Expression) for elem in fields_or_schema)):
self._j_tenv.createTemporaryView(
view_path,
j_data_stream,
to_expression_jarray(fields_or_schema))
else:
raise ValueError("Invalid arguments for 'fields': %r" %
','.join([repr(item) for item in fields_or_schema]))
def create_view(self,
view_path: str,
table: Table,
ignore_if_exists: Optional[bool] = False):
"""
Registers a :class:`~pyflink.table.Table` API object as a view similar to SQL views.
This method can only create permanent objects. Temporary objects can shadow permanent ones.
If a temporary object exists in a given path,
make sure to drop the temporary object first using :func:`drop_temporary_view`.
:param view_path: The path under which the view will be registered.
See also the :class:`~pyflink.table.TableEnvironment` class description
for the format of the path.
:param table: The view to register.
:param ignore_if_exists: If a view or a table exists and the given flag is set,
no operation is executed. An exception is thrown otherwise.
.. versionadded:: 2.0.0
"""
self._j_tenv.createView(view_path, table, ignore_if_exists)
def create_model(self,
model_path: str,
model_descriptor: ModelDescriptor,
ignore_if_exists: Optional[bool] = False):
"""
Registers the given :class:`~pyflink.table.ModelDescriptor` as a catalog model
similar to SQL models.
The ModelDescriptor is converted into a CatalogModel and stored in the catalog.
If the model should not be permanently stored in a catalog, use
:func:`create_temporary_model` instead.
Examples:
::
>>> table_env.create_model("MyModel", ModelDescriptor.for_provider("OPENAI")
... .input_schema(Schema.new_builder()
... .column("f0", DataTypes.STRING())
... .build())
... .output_schema(Schema.new_builder()
... .column("label", DataTypes.STRING())
... .build())
... .option("task", "regression")
... .option("type", "remote")
... .
... .
... .build(),
... True)
:param model_path: The path under which the model will be registered.
:param model_descriptor: Template for creating a CatalogModel instance.
:param ignore_if_exists: If a model exists under the given path and this flag is set,
no operation is executed. An exception is thrown otherwise.
.. versionadded:: 2.1.0
"""
self._j_tenv.createModel(model_path, model_descriptor._j_model_descriptor, ignore_if_exists)
@PublicEvolving()
def from_model(self, model_path: Union[str, ModelDescriptor]) -> 'Model':
"""
Gets a registered model or creates an anonymous model from a descriptor.
A descriptor defines the provider options and input/output schemas directly,
without registering a model in the catalog.
Example:
::
>>> from pyflink.table import ModelDescriptor, Schema
>>> model = t_env.from_model("my_model")
>>> input_table = t_env.from_path("src")
>>> result_table = model.predict(input_table, ["user_comment"])
Create an anonymous model:
::
>>> from pyflink.table import DataTypes
>>> descriptor = (ModelDescriptor.for_provider("my-provider")
... .input_schema(Schema.new_builder()
... .column("text", DataTypes.STRING()).build())
... .output_schema(Schema.new_builder()
... .column("label", DataTypes.STRING()).build())
... .build())
>>> model = t_env.from_model(descriptor)
:param model_path: The path of the registered model, or a descriptor defining
an anonymous model.
:return: The requested model instance.
.. versionadded:: 2.4.0
"""
if isinstance(model_path, ModelDescriptor):
j_model = self._j_tenv.fromModel(model_path._j_model_descriptor)
else:
j_model = self._j_tenv.fromModel(model_path)
return Model(j_model)
def create_temporary_model(self,
model_path: str,
model_descriptor: ModelDescriptor,
ignore_if_exists: Optional[bool] = False):
"""
Registers the given :class:`~pyflink.table.ModelDescriptor` as a temporary catalog model
similar to SQL temporary models.
The ModelDescriptor is converted into a CatalogModel and stored in the catalog.
Temporary objects can shadow permanent ones. If a permanent object in a given path
exists, it will be inaccessible in the current session. To make the permanent object
available again one can drop the corresponding temporary object.
Examples:
::
>>> table_env.create_temporary_model("MyModel", ModelDescriptor
... .for_provider("OPENAI")
... .input_schema(Schema.new_builder()
... .column("f0", DataTypes.STRING())
... .build())
... .output_schema(Schema.new_builder()
... .column("label", DataTypes.STRING())
... .build())
... .option("task", "regression")
... .option("type", "remote")
... .
... .
... .build(),
... True)
:param model_path: The path under which the model will be registered.
:param model_descriptor: Template for creating a CatalogModel instance.
:param ignore_if_exists: If a model exists under the given path and this flag is
set, no operation is executed. An exception is thrown
otherwise.
.. versionadded:: 2.1.0
"""
self._j_tenv.createTemporaryModel(model_path, model_descriptor._j_model_descriptor,
ignore_if_exists)
def add_python_file(self, file_path: str):
"""
Adds a python dependency which could be python files, python packages or
local directories. They will be added to the PYTHONPATH of the python UDF worker.
Please make sure that these dependencies can be imported.
:param file_path: The path of the python dependency.
.. versionadded:: 1.10.0
"""
jvm = get_gateway().jvm
python_files = self.get_config().get(jvm.PythonOptions.PYTHON_FILES.key(), None)
if python_files is not None:
python_files = jvm.PythonDependencyUtils.FILE_DELIMITER.join([file_path, python_files])
else:
python_files = file_path
self.get_config().set(jvm.PythonOptions.PYTHON_FILES.key(), python_files)
def set_python_requirements(self,
requirements_file_path: str,
requirements_cache_dir: str = None):
"""
Specifies a requirements.txt file which defines the third-party dependencies.
These dependencies will be installed to a temporary directory and added to the
PYTHONPATH of the python UDF worker.
For the dependencies which could not be accessed in the cluster, a directory which contains
the installation packages of these dependencies could be specified using the parameter
"requirements_cached_dir". It will be uploaded to the cluster to support offline
installation.
Example:
::
# commands executed in shell
$ echo numpy==1.16.5 > requirements.txt
$ pip download -d cached_dir -r requirements.txt --no-binary :all:
# python code
>>> table_env.set_python_requirements("requirements.txt", "cached_dir")
.. note::
Please make sure the installation packages matches the platform of the cluster
and the python version used. These packages will be installed using pip,
so also make sure the version of Pip (version >= 20.3) and the version of
SetupTools (version >= 37.0.0).
:param requirements_file_path: The path of "requirements.txt" file.
:param requirements_cache_dir: The path of the local directory which contains the
installation packages.
.. versionadded:: 1.10.0
"""
jvm = get_gateway().jvm
python_requirements = requirements_file_path
if requirements_cache_dir is not None:
python_requirements = jvm.PythonDependencyUtils.PARAM_DELIMITER.join(
[python_requirements, requirements_cache_dir])
self.get_config().set(
jvm.PythonOptions.PYTHON_REQUIREMENTS.key(), python_requirements)
def add_python_archive(self, archive_path: str, target_dir: str = None):
"""
Adds a python archive file. The file will be extracted to the working directory of
python UDF worker.
If the parameter "target_dir" is specified, the archive file will be extracted to a
directory named ${target_dir}. Otherwise, the archive file will be extracted to a
directory with the same name of the archive file.
If python UDF depends on a specific python version which does not exist in the cluster,
this method can be used to upload the virtual environment.
Note that the path of the python interpreter contained in the uploaded environment
should be specified via the method :func:`pyflink.table.TableConfig.set_python_executable`.
The files uploaded via this method are also accessible in UDFs via relative path.
Example:
::
# command executed in shell
# assert the relative path of python interpreter is py_env/bin/python
$ zip -r py_env.zip py_env
# python code
>>> table_env.add_python_archive("py_env.zip")
>>> table_env.get_config().set_python_executable("py_env.zip/py_env/bin/python")
# or
>>> table_env.add_python_archive("py_env.zip", "myenv")
>>> table_env.get_config().set_python_executable("myenv/py_env/bin/python")
# the files contained in the archive file can be accessed in UDF
>>> def my_udf():
... with open("myenv/py_env/data/data.txt") as f:
... ...
.. note::
Please make sure the uploaded python environment matches the platform that the cluster
is running on and that the python version must be 3.5 or higher.
.. note::
Currently only zip-format is supported. i.e. zip, jar, whl, egg, etc.
The other archive formats such as tar, tar.gz, 7z, rar, etc are not supported.
:param archive_path: The archive file path.
:param target_dir: Optional, the target dir name that the archive file extracted to.
.. versionadded:: 1.10.0
"""
jvm = get_gateway().jvm
if target_dir is not None:
archive_path = jvm.PythonDependencyUtils.PARAM_DELIMITER.join(
[archive_path, target_dir])
python_archives = self.get_config().get(jvm.PythonOptions.PYTHON_ARCHIVES.key(), None)
if python_archives is not None:
python_files = jvm.PythonDependencyUtils.FILE_DELIMITER.join(
[python_archives, archive_path])
else:
python_files = archive_path
self.get_config().set(jvm.PythonOptions.PYTHON_ARCHIVES.key(), python_files)
def from_elements(self, elements: Iterable, schema: Union[DataType, List[str]] = None,
verify_schema: bool = True) -> Table:
"""
Creates a table from a collection of elements.
The elements types must be acceptable atomic types or acceptable composite types.
All elements must be of the same type.
If the elements types are composite types, the composite types must be strictly equal,
and its subtypes must also be acceptable types.
e.g. if the elements are tuples, the length of the tuples must be equal, the element types
of the tuples must be equal in order.
The built-in acceptable atomic element types contains:
**int**, **long**, **str**, **unicode**, **bool**,
**float**, **bytearray**, **datetime.date**, **datetime.time**, **datetime.datetime**,
**datetime.timedelta**, **decimal.Decimal**
The built-in acceptable composite element types contains:
**list**, **tuple**, **dict**, **array**, :class:`~pyflink.table.Row`
If the element type is a composite type, it will be unboxed.
e.g. table_env.from_elements([(1, 'Hi'), (2, 'Hello')]) will return a table like:
+----+-------+
| _1 | _2 |
+====+=======+
| 1 | Hi |
+----+-------+
| 2 | Hello |
+----+-------+
"_1" and "_2" are generated field names.
Example:
::
# use the second parameter to specify custom field names
>>> table_env.from_elements([(1, 'Hi'), (2, 'Hello')], ['a', 'b'])
# use the second parameter to specify custom table schema
>>> table_env.from_elements([(1, 'Hi'), (2, 'Hello')],
... DataTypes.ROW([DataTypes.FIELD("a", DataTypes.INT()),
... DataTypes.FIELD("b", DataTypes.STRING())]))
# use the third parameter to switch whether to verify the elements against the schema
>>> table_env.from_elements([(1, 'Hi'), (2, 'Hello')],
... DataTypes.ROW([DataTypes.FIELD("a", DataTypes.INT()),
... DataTypes.FIELD("b", DataTypes.STRING())]),
... False)
# create Table from expressions
>>> table_env.from_elements([row(1, 'abc', 2.0), row(2, 'def', 3.0)],
... DataTypes.ROW([DataTypes.FIELD("a", DataTypes.INT()),
... DataTypes.FIELD("b", DataTypes.STRING()),
... DataTypes.FIELD("c", DataTypes.FLOAT())]))
: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: The result table.
"""
# verifies the elements against the specified schema
if isinstance(schema, RowType):
verify_func = _create_type_verifier(schema) if verify_schema else lambda _: True
def verify_obj(obj):
verify_func(obj)
return obj
elif isinstance(schema, DataType):
data_type = schema
schema = RowType().add("value", schema)
verify_func = _create_type_verifier(
data_type, name="field value") if verify_schema else lambda _: True
def verify_obj(obj):
verify_func(obj)
return obj
else:
def verify_obj(obj):
return obj
# infers the schema if not specified
if schema is None or isinstance(schema, (list, tuple)):
schema = _infer_schema_from_data(elements, names=schema)
converter = _create_converter(schema)
elements = map(converter, elements)
elif not isinstance(schema, RowType):
raise TypeError(
"schema should be RowType, list, tuple or None, but got: %s" % schema)
elements = list(elements)
# in case all the elements are expressions
if len(elements) > 0 and all(isinstance(elem, Expression) for elem in elements):
if schema is None:
return Table(self._j_tenv.fromValues(to_expression_jarray(elements)), self)
else:
return Table(self._j_tenv.fromValues(_to_java_data_type(schema),
to_expression_jarray(elements)),
self)
elif any(isinstance(elem, Expression) for elem in elements):
raise ValueError("It doesn't support part of the elements are Expression, while the "
"others are not.")
# verifies the elements against the specified schema
elements = map(verify_obj, elements)
# converts python data to sql data
elements = [schema.to_sql_type(element) for element in elements]
return self._from_elements(elements, schema)
def _from_elements(
self,
elements: List,
schema: DataType,
table_schema: Schema = None) -> Table:
"""
Creates a table from a collection of elements.
:param elements: The elements to create a table from.
:param schema: Data type used to serialize the elements.
:param table_schema: Optional declarative schema for the resulting source table.
:return: The result :class:`~pyflink.table.Table`.
"""
# serializes to a file, and we read the file in java
temp_file = tempfile.NamedTemporaryFile(delete=False, dir=tempfile.mkdtemp())
serializer = BatchedSerializer(self._serializer)
try:
with temp_file:
serializer.serialize(elements, temp_file)
j_schema = (table_schema._j_schema if table_schema is not None
else _to_java_data_type(schema))
gateway = get_gateway()
PythonTableUtils = gateway.jvm \
.org.apache.flink.table.utils.python.PythonTableUtils
j_table = PythonTableUtils.createTableFromElement(
self._j_tenv, temp_file.name, j_schema, True)
return Table(j_table, self)
finally:
atexit.register(lambda: os.unlink(temp_file.name))
def _from_arrow(
self,
table,
row_type: RowType,
table_schema: Schema = None,
splits_num: int = 1) -> Table:
"""Creates a table from a PyArrow Table through the Arrow table source."""
import pyarrow as pa
if not isinstance(table, pa.Table):
raise TypeError(f"table must be a pyarrow.Table, but was {type(table).__name__}")
if isinstance(splits_num, bool) or not isinstance(splits_num, int):
raise TypeError("splits_num must be an integer")
if splits_num <= 0:
raise ValueError("splits_num must be greater than 0")
field_names = row_type.field_names()
field_types = row_type.field_types()
try:
compatible_table = table.rename_columns(field_names)
for index, (field, data_type) in enumerate(zip(table.schema, field_types)):
source_column = compatible_table.column(index)
column = source_column
target_type = column.type
if isinstance(data_type, TimestampType):
target_type = to_arrow_type(data_type)
if column.type.tz is not None:
target_type = pa.timestamp(
target_type.unit, tz=column.type.tz)
if column.type != target_type:
if pa.types.is_timestamp(column.type):
column = _cast_arrow_timestamp(column, target_type)
else:
column = column.cast(target_type, safe=False)
if column is source_column:
continue
target_field = pa.field(
field_names[index],
target_type,
field.nullable,
field.metadata,
)
compatible_table = compatible_table.set_column(
index,
target_field,
column,
)
except (
pa.ArrowInvalid,
pa.ArrowNotImplementedError,
pa.ArrowTypeError,
TypeError,
ValueError,
) as e:
raise TypeError(
f"Could not convert pyarrow.Table to the inferred Flink schema: {row_type}"
) from e
with tempfile.TemporaryDirectory() as temp_dir:
with tempfile.NamedTemporaryFile(delete=False, dir=temp_dir) as temp_file:
with pa.ipc.new_stream(temp_file, compatible_table.schema) as writer:
if compatible_table.num_rows > 0:
max_chunksize = -(-compatible_table.num_rows // splits_num)
writer.write_table(
compatible_table, max_chunksize=max_chunksize)
jvm = get_gateway().jvm
if table_schema is None:
source_schema = _to_java_data_type(row_type).notNull()
source_schema = source_schema.bridgedTo(
load_java_class('org.apache.flink.table.data.RowData'))
else:
source_schema = table_schema._j_schema
create_descriptor = jvm.org.apache.flink.table.runtime.arrow.ArrowUtils \
.createArrowTableSourceDesc
if any(
_contains_timezone_aware_timestamp(field.type)
for field in compatible_table.schema
):
descriptor = create_descriptor(
source_schema,
temp_file.name,
self.get_config().get_local_timezone(),
)
else:
descriptor = create_descriptor(source_schema, temp_file.name)
return Table(getattr(self._j_tenv, "from")(descriptor), self)
def from_pandas(self, pdf: 'pandas.DataFrame',
schema: Union[RowType, List[str], Tuple[str], List[DataType],
Tuple[DataType]] = None,
splits_num: int = 1) -> Table:
"""
Creates a table from a pandas DataFrame.
Example:
::
>>> pdf = pd.DataFrame(np.random.rand(1000, 2))
# use the second parameter to specify custom field names
>>> table_env.from_pandas(pdf, ["a", "b"])
# use the second parameter to specify custom field types
>>> table_env.from_pandas(pdf, [DataTypes.DOUBLE(), DataTypes.DOUBLE()]))
# use the second parameter to specify custom table schema
>>> table_env.from_pandas(pdf,
... DataTypes.ROW([DataTypes.FIELD("a", DataTypes.DOUBLE()),
... DataTypes.FIELD("b", DataTypes.DOUBLE())]))
:param pdf: The pandas DataFrame.
:param schema: The schema of the converted table.
:param splits_num: The number of splits the given Pandas DataFrame will be split into. It
determines the number of parallel source tasks.
If not specified, the default parallelism will be used.
:return: The result table.
.. versionadded:: 1.11.0
"""
import pandas as pd
if not isinstance(pdf, pd.DataFrame):
raise TypeError("Unsupported type, expected pandas.DataFrame, got %s" % type(pdf))
import pyarrow as pa
arrow_schema = pa.Schema.from_pandas(pdf, preserve_index=False)
if schema is not None:
if isinstance(schema, RowType):
result_type = schema
elif isinstance(schema, (list, tuple)) and isinstance(schema[0], str):
result_type = RowType(
[RowField(field_name, from_arrow_type(field.type, field.nullable))
for field_name, field in zip(schema, arrow_schema)])
elif isinstance(schema, (list, tuple)) and isinstance(schema[0], DataType):
result_type = RowType(
[RowField(field_name, field_type) for field_name, field_type in zip(
arrow_schema.names, schema)])
else:
raise TypeError("Unsupported schema type, it could only be of RowType, a "
"list of str or a list of DataType, got %s" % schema)
else:
result_type = RowType([RowField(field.name, from_arrow_type(field.type, field.nullable))
for field in arrow_schema])
# serializes to a file, and we read the file in java
temp_file = tempfile.NamedTemporaryFile(delete=False, dir=tempfile.mkdtemp())
import pytz
serializer = ArrowSerializer(
create_arrow_schema(result_type.field_names(), result_type.field_types()),
result_type,
pytz.timezone(self.get_config().get_local_timezone()))
step = -(-len(pdf) // splits_num)
pdf_slices = [pdf.iloc[start:start + step] for start in range(0, len(pdf), step)]
data = [[c for (_, c) in pdf_slice.items()] for pdf_slice in pdf_slices]
try:
with temp_file:
serializer.serialize(data, temp_file)
jvm = get_gateway().jvm
data_type = _to_java_data_type(result_type).notNull()
data_type = data_type.bridgedTo(
load_java_class('org.apache.flink.table.data.RowData'))
j_arrow_table_source_descriptor = \
jvm.org.apache.flink.table.runtime.arrow.ArrowUtils.createArrowTableSourceDesc(
data_type, temp_file.name)
return Table(getattr(self._j_tenv, "from")(j_arrow_table_source_descriptor), self)
finally:
os.unlink(temp_file.name)
def load_plan(self, plan_reference: PlanReference) -> CompiledPlan:
"""
Loads a plan from a :class:`~pyflink.table.PlanReference` into a
:class:`~pyflink.table.CompiledPlan`.
Compiled plans can be persisted and reloaded across Flink versions. They describe static
pipelines to ensure backwards compatibility and enable stateful streaming job upgrades. See
:class:`~pyflink.table.CompiledPlan` and the website documentation for more information.
This method will parse the input reference and will validate the plan. The returned
instance can be executed via :func:`~pyflink.table.CompiledPlan.execute`.
.. note::
The compiled plan feature is experimental in batch mode.
:raises TableException: if the plan cannot be loaded from the filesystem, or if the plan
is invalid.
.. versionadded:: 2.1.0
"""
return CompiledPlan(
j_compiled_plan=self._j_tenv.loadPlan(plan_reference._j_plan_reference),
t_env=self
)
def compile_plan_sql(self, stmt: str) -> CompiledPlan:
"""
Compiles a SQL DML statement into a :class:`~pyflink.table.CompiledPlan`.
Compiled plans can be persisted and reloaded across Flink versions. They describe static
pipelines to ensure backwards compatibility and enable stateful streaming job upgrades. See
:class:`~pyflink.table.CompiledPlan` and the website documentation for more information.
.. note::
Only ``INSERT INTO`` is supported at the moment.
.. note::
The compiled plan feature is experimental in batch mode.
.. seealso::
:func:`~pyflink.table.TableEnvironment.load_plan`
:func:`~pyflink.table.CompiledPlan.execute`
:raises TableException: if the SQL statement is invalid or if the plan cannot be
persisted.
.. versionadded:: 2.1.0
"""
return CompiledPlan(j_compiled_plan=self._j_tenv.compilePlanSql(stmt), t_env=self)
def execute_plan(self, plan_reference: PlanReference) -> TableResult:
"""
Shorthand for ``tEnv.load_plan(plan_reference).execute()``.
.. seealso::
:func:`~pyflink.table.TableEnvironment.load_plan`
:func:`~pyflink.table.CompiledPlan.execute`
.. versionadded:: 2.1.0
"""
return self.load_plan(plan_reference).execute()
def _set_python_executable_for_local_executor(self):
jvm = get_gateway().jvm
j_config = get_j_env_configuration(self._get_j_env())
if not j_config.containsKey(jvm.PythonOptions.PYTHON_EXECUTABLE.key()) \
and is_local_deployment(j_config):
j_config.setString(jvm.PythonOptions.PYTHON_EXECUTABLE.key(), sys.executable)
def _add_jars_to_j_env_config(self, config_key):
jar_urls = self.get_config().get(config_key, None)
if jar_urls:
jvm = get_gateway().jvm
jar_urls_list = []
parsed_jar_urls = Configuration.parse_list_value(jar_urls)
url_strings = [
jvm.java.net.URL(url).toString() if url else ""
for url in parsed_jar_urls
]
self._parse_urls(url_strings, jar_urls_list)
j_configuration = get_j_env_configuration(self._get_j_env())
parsed_jar_urls = Configuration.parse_list_value(
j_configuration.getString(config_key, "")
)
self._parse_urls(parsed_jar_urls, jar_urls_list)
j_configuration.setString(config_key, ";".join(jar_urls_list))
def _parse_urls(self, jar_urls, jar_urls_list):
for url in jar_urls:
url = url.strip()
if url != "" and url not in jar_urls_list:
jar_urls_list.append(url)
def _get_j_env(self):
return self._j_tenv.getPlanner().getExecEnv()
@staticmethod
def _is_table_function(java_function):
java_function_class = java_function.getClass()
j_table_function_class = get_java_class(
get_gateway().jvm.org.apache.flink.table.functions.TableFunction)
return j_table_function_class.isAssignableFrom(java_function_class)
@staticmethod
def _is_aggregate_function(java_function):
java_function_class = java_function.getClass()
j_aggregate_function_class = get_java_class(
get_gateway().jvm.org.apache.flink.table.functions.ImperativeAggregateFunction)
return j_aggregate_function_class.isAssignableFrom(java_function_class)
def _register_table_function(self, name, table_function):
function_catalog = self._get_function_catalog()
gateway = get_gateway()
helper = gateway.jvm.org.apache.flink.table.functions.UserDefinedFunctionHelper
result_type = helper.getReturnTypeOfTableFunction(table_function)
function_catalog.registerTempSystemTableFunction(name, table_function, result_type)
def _register_aggregate_function(self, name, aggregate_function):
function_catalog = self._get_function_catalog()
gateway = get_gateway()
helper = gateway.jvm.org.apache.flink.table.functions.UserDefinedFunctionHelper
result_type = helper.getReturnTypeOfAggregateFunction(aggregate_function)
acc_type = helper.getAccumulatorTypeOfAggregateFunction(aggregate_function)
function_catalog.registerTempSystemAggregateFunction(
name, aggregate_function, result_type, acc_type)
def _get_function_catalog(self):
function_catalog_field = self._j_tenv.getClass().getDeclaredField("functionCatalog")
function_catalog_field.setAccessible(True)
function_catalog = function_catalog_field.get(self._j_tenv)
return function_catalog
def _before_execute(self):
jvm = get_gateway().jvm
jars_key = jvm.org.apache.flink.configuration.PipelineOptions.JARS.key()
classpaths_key = jvm.org.apache.flink.configuration.PipelineOptions.CLASSPATHS.key()
self._add_jars_to_j_env_config(jars_key)
self._add_jars_to_j_env_config(classpaths_key)
def _wrap_aggregate_function_if_needed(self, function) -> UserDefinedFunctionWrapper:
if isinstance(function, AggregateFunction):
function = udaf(function,
result_type=function.get_result_type(),
accumulator_type=function.get_accumulator_type(),
name=str(function.__class__.__name__))
elif isinstance(function, TableAggregateFunction):
function = udtaf(function,
result_type=function.get_result_type(),
accumulator_type=function.get_accumulator_type(),
name=str(function.__class__.__name__))
return function
def _config_chaining_optimization(self):
JChainingOptimizingExecutor = get_gateway().jvm.org.apache.flink.table.executor.python.\
ChainingOptimizingExecutor
exec_env_field = get_field(self._j_tenv.getClass(), "execEnv")
exec_env_field.set(self._j_tenv,
JChainingOptimizingExecutor(exec_env_field.get(self._j_tenv)))
def _open(self):
# start BeamFnLoopbackWorkerPoolServicer when executed in MiniCluster
def startup_loopback_server():
from pyflink.fn_execution.beam.beam_worker_pool_service import \
BeamFnLoopbackWorkerPoolServicer
self.get_config().set("python.loopback-server.address",
BeamFnLoopbackWorkerPoolServicer()
.configure(
Configuration.from_dict(self.get_config().to_map())).start())
python_worker_execution_mode = os.environ.get('_python_worker_execution_mode')
if python_worker_execution_mode is None:
if is_local_deployment(get_j_env_configuration(self._get_j_env())):
startup_loopback_server()
elif python_worker_execution_mode == 'loopback':
if is_local_deployment(get_j_env_configuration(self._get_j_env())):
startup_loopback_server()
else:
raise ValueError("Loopback mode is enabled, however the job wasn't configured to "
"run in local deployment mode")
elif python_worker_execution_mode != 'process':
raise ValueError(
"It only supports to execute the Python worker in 'loopback' mode and 'process' "
"mode, unknown mode '%s' is configured" % python_worker_execution_mode)
class StreamTableEnvironment(TableEnvironment):
def __init__(self, j_tenv):
super(StreamTableEnvironment, self).__init__(j_tenv)
@staticmethod
def create(stream_execution_environment: StreamExecutionEnvironment = None, # type: ignore
environment_settings: EnvironmentSettings = None) -> 'StreamTableEnvironment':
"""
Creates a :class:`~pyflink.table.StreamTableEnvironment`.
Example:
::
# create with StreamExecutionEnvironment.
>>> env = StreamExecutionEnvironment.get_execution_environment()
>>> table_env = StreamTableEnvironment.create(env)
# create with StreamExecutionEnvironment and EnvironmentSettings.
>>> configuration = Configuration()
>>> configuration.set_string('execution.buffer-timeout', '1 min')
>>> environment_settings = EnvironmentSettings \\
... .new_instance() \\
... .in_streaming_mode() \\
... .with_configuration(configuration) \\
... .build()
>>> table_env = StreamTableEnvironment.create(
... env, environment_settings=environment_settings)
# create with EnvironmentSettings.
>>> table_env = StreamTableEnvironment.create(environment_settings=environment_settings)
:param stream_execution_environment: The
:class:`~pyflink.datastream.StreamExecutionEnvironment`
of the TableEnvironment.
:param environment_settings: The environment settings used to instantiate the
TableEnvironment.
:return: The StreamTableEnvironment created from given StreamExecutionEnvironment and
configuration.
"""
if stream_execution_environment is None and \
environment_settings is None:
raise ValueError("No argument found, the param 'stream_execution_environment' "
"or 'environment_settings' is required.")
gateway = get_gateway()
if environment_settings is not None:
if stream_execution_environment is None:
j_tenv = gateway.jvm.TableEnvironment.create(
environment_settings._j_environment_settings)
else:
j_tenv = gateway.jvm.StreamTableEnvironment.create(
stream_execution_environment._j_stream_execution_environment,
environment_settings._j_environment_settings)
else:
j_tenv = gateway.jvm.StreamTableEnvironment.create(
stream_execution_environment._j_stream_execution_environment)
return StreamTableEnvironment(j_tenv)
def from_data_stream(self,
data_stream: DataStream,
*fields_or_schema: Union[Expression, Schema]) -> Table:
"""
1. When fields_or_schema is a sequence of Expression:
Converts the given DataStream into a Table with specified field names.
There are two modes for mapping original fields to the fields of the Table:
1. Reference input fields by name:
All fields in the schema definition are referenced by name (and possibly renamed
using and alias (as). Moreover, we can define proctime and rowtime attributes at
arbitrary positions using arbitrary names (except those that exist in the result
schema). In this mode, fields can be reordered and projected out. This mode can be
used for any input type.
2. Reference input fields by position:
In this mode, fields are simply renamed. Event-time attributes can replace the field
on their position in the input data (if it is of correct type) or be appended at the
end. Proctime attributes must be appended at the end. This mode can only be used if
the input type has a defined field order (tuple, case class, Row) and none of the
fields references a field of the input type.
2. When fields_or_schema is a Schema:
Converts the given DataStream into a Table.
Column names and types of the Table are automatically derived from the TypeInformation
of the DataStream. If the outermost record's TypeInformation is a CompositeType, it will
be flattened in the first level. Composite nested fields will not be accessible.
Since the DataStream API does not support changelog processing natively, this method
assumes append-only/insert-only semantics during the stream-to-table conversion. Records
of class Row must describe RowKind.INSERT changes.
By default, the stream record's timestamp and watermarks are not propagated unless
explicitly declared.
This method allows to declare a Schema for the resulting table. The declaration is
similar to a {@code CREATE TABLE} DDL in SQL and allows to:
1. enrich or overwrite automatically derived columns with a custom DataType
2. reorder columns
3. add computed or metadata columns next to the physical columns
4. access a stream record's timestamp
5. declare a watermark strategy or propagate the DataStream watermarks
It is possible to declare a schema without physical/regular columns. In this case, those
columns will be automatically derived and implicitly put at the beginning of the schema
declaration.
The following examples illustrate common schema declarations and their semantics:
Example:
::
=== EXAMPLE 1 ===
no physical columns defined, they will be derived automatically,
e.g. BigDecimal becomes DECIMAL(38, 18)
>>> Schema.new_builder() \
... .column_by_expression("c1", "f1 + 42") \
... .column_by_expression("c2", "f1 - 1") \
... .build()
equal to: CREATE TABLE (f0 STRING, f1 DECIMAL(38, 18), c1 AS f1 + 42, c2 AS f1 - 1)
=== EXAMPLE 2 ===
physical columns defined, input fields and columns will be mapped by name,
columns are reordered and their data type overwritten,
all columns must be defined to show up in the final table's schema
>>> Schema.new_builder() \
... .column("f1", "DECIMAL(10, 2)") \
... .column_by_expression("c", "f1 - 1") \
... .column("f0", "STRING") \
... .build()
equal to: CREATE TABLE (f1 DECIMAL(10, 2), c AS f1 - 1, f0 STRING)
=== EXAMPLE 3 ===
timestamp and watermarks can be added from the DataStream API,
physical columns will be derived automatically
>>> Schema.new_builder() \
... .column_by_metadata("rowtime", "TIMESTAMP_LTZ(3)") \
... .watermark("rowtime", "SOURCE_WATERMARK()") \
... .build()
equal to:
CREATE TABLE (
f0 STRING,
f1 DECIMAL(38, 18),
rowtime TIMESTAMP(3) METADATA,
WATERMARK FOR rowtime AS SOURCE_WATERMARK()
)
.. note:: create_temporary_view by providing a Schema (case 2.) was added from flink
1.14.0.
:param data_stream: The datastream to be converted.
:param fields_or_schema: The fields expressions to map original fields of the DataStream to
the fields of the Table or the customized schema for the final table.
:return: The converted Table.
.. versionadded:: 1.12.0
"""
j_data_stream = data_stream._j_data_stream
JPythonConfigUtil = get_gateway().jvm.org.apache.flink.python.util.PythonConfigUtil
JPythonConfigUtil.configPythonOperator(j_data_stream.getExecutionEnvironment())
if len(fields_or_schema) == 0:
return Table(j_table=self._j_tenv.fromDataStream(j_data_stream), t_env=self)
elif all(isinstance(f, Expression) for f in fields_or_schema):
return Table(j_table=self._j_tenv.fromDataStream(
j_data_stream, to_expression_jarray(fields_or_schema)), t_env=self)
elif len(fields_or_schema) == 1 and isinstance(fields_or_schema[0], Schema):
return Table(j_table=self._j_tenv.fromDataStream(
j_data_stream, fields_or_schema[0]._j_schema), t_env=self)
raise ValueError("Invalid arguments for 'fields': %r" % fields_or_schema)
def from_changelog_stream(self,
data_stream: DataStream,
schema: Schema = None,
changelog_mode: ChangelogMode = None) -> Table:
"""
Converts the given DataStream of changelog entries into a Table.
Compared to :func:`from_data_stream`, this method consumes instances of Row and evaluates
the RowKind flag that is contained in every record during runtime. The runtime behavior is
similar to that of a DynamicTableSource.
If you don't specify the changelog_mode, the changelog containing all kinds of changes
(enumerated in RowKind) as the default ChangelogMode.
Column names and types of the Table are automatically derived from the TypeInformation of
the DataStream. If the outermost record's TypeInformation is a CompositeType, it will be
flattened in the first level. Composite nested fields will not be accessible.
By default, the stream record's timestamp and watermarks are not propagated unless
explicitly declared.
This method allows to declare a Schema for the resulting table. The declaration is similar
to a {@code CREATE TABLE} DDL in SQL and allows to:
1. enrich or overwrite automatically derived columns with a custom DataType
2. reorder columns
3. add computed or metadata columns next to the physical columns
4. access a stream record's timestamp
5. declare a watermark strategy or propagate the DataStream watermarks
6. declare a primary key
See :func:`from_data_stream` for more information and examples of how to declare a Schema.
:param data_stream: The changelog stream of Row.
:param schema: The customized schema for the final table.
:param changelog_mode: The expected kinds of changes in the incoming changelog.
:return: The converted Table.
"""
j_data_stream = data_stream._j_data_stream
JPythonConfigUtil = get_gateway().jvm.org.apache.flink.python.util.PythonConfigUtil
JPythonConfigUtil.configPythonOperator(j_data_stream.getExecutionEnvironment())
if schema is None:
return Table(self._j_tenv.fromChangelogStream(j_data_stream), t_env=self)
elif changelog_mode is None:
return Table(
self._j_tenv.fromChangelogStream(j_data_stream, schema._j_schema), t_env=self)
else:
return Table(
self._j_tenv.fromChangelogStream(
j_data_stream,
schema._j_schema,
changelog_mode._j_changelog_mode),
t_env=self)
def to_data_stream(self, table: Table) -> DataStream:
"""
Converts the given Table into a DataStream.
Since the DataStream API does not support changelog processing natively, this method
assumes append-only/insert-only semantics during the table-to-stream conversion. The records
of class Row will always describe RowKind#INSERT changes. Updating tables are
not supported by this method and will produce an exception.
Note that the type system of the table ecosystem is richer than the one of the DataStream
API. The table runtime will make sure to properly serialize the output records to the first
operator of the DataStream API. Afterwards, the Types semantics of the DataStream API
need to be considered.
If the input table contains a single rowtime column, it will be propagated into a stream
record's timestamp. Watermarks will be propagated as well.
:param table: The Table to convert.
:return: The converted DataStream.
"""
return DataStream(self._j_tenv.toDataStream(table._j_table))
def to_changelog_stream(self,
table: Table,
target_schema: Schema = None,
changelog_mode: ChangelogMode = None) -> DataStream:
"""
Converts the given Table into a DataStream of changelog entries.
Compared to :func:`to_data_stream`, this method produces instances of Row and sets the
RowKind flag that is contained in every record during runtime. The runtime behavior is
similar to that of a DynamicTableSink.
If you don't specify the changelog_mode, the changelog containing all kinds of changes
(enumerated in RowKind) as the default ChangelogMode.
The given Schema is used to configure the table runtime to convert columns and internal data
structures to the desired representation. The following example shows how to
convert a table column into a Row type.
Example:
::
>>> table_env.to_changelog_stream(
... table,
... Schema.new_builder() \
... .column("id", DataTypes.BIGINT())
... .column("payload", DataTypes.ROW(
... [DataTypes.FIELD("name", DataTypes.STRING()),
... DataTypes.FIELD("age", DataTypes.INT())]))
... .build())
Note that the type system of the table ecosystem is richer than the one of the DataStream
API. The table runtime will make sure to properly serialize the output records to the first
operator of the DataStream API. Afterwards, the Types semantics of the DataStream API need
to be considered.
If the input table contains a single rowtime column, it will be propagated into a stream
record's timestamp. Watermarks will be propagated as well.
If the rowtime should not be a concrete field in the final Row anymore, or the schema should
be symmetrical for both :func:`from_changelog_stream` and :func:`to_changelog_stream`, the
rowtime can also be declared as a metadata column that will be propagated into a stream
record's timestamp. It is possible to declare a schema without physical/regular columns.
In this case, those columns will be automatically derived and implicitly put at the
beginning of the schema declaration.
The following examples illustrate common schema declarations and their semantics:
Example:
::
given a Table of (id INT, name STRING, my_rowtime TIMESTAMP_LTZ(3))
=== EXAMPLE 1 ===
no physical columns defined, they will be derived automatically,
the last derived physical column will be skipped in favor of the metadata column
>>> Schema.new_builder() \
... .column_by_metadata("rowtime", "TIMESTAMP_LTZ(3)") \
... .build()
equal to: CREATE TABLE (id INT, name STRING, rowtime TIMESTAMP_LTZ(3) METADATA)
=== EXAMPLE 2 ===
physical columns defined, all columns must be defined
>>> Schema.new_builder() \
... .column("id", "INT") \
... .column("name", "STRING") \
... .column_by_metadata("rowtime", "TIMESTAMP_LTZ(3)") \
... .build()
equal to: CREATE TABLE (id INT, name STRING, rowtime TIMESTAMP_LTZ(3) METADATA)
:param table: The Table to convert. It can be updating or insert-only.
:param target_schema: The Schema that decides about the final external representation in
DataStream records.
:param changelog_mode: The required kinds of changes in the result changelog. An exception
will be thrown if the given updating table cannot be represented in this changelog mode.
:return: The converted changelog stream of Row.
"""
if target_schema is None:
return DataStream(self._j_tenv.toChangelogStream(table._j_table))
elif changelog_mode is None:
return DataStream(
self._j_tenv.toChangelogStream(table._j_table, target_schema._j_schema))
else:
return DataStream(
self._j_tenv.toChangelogStream(
table._j_table,
target_schema._j_schema,
changelog_mode._j_changelog_mode))
def to_append_stream(self, table: Table, type_info: TypeInformation) -> DataStream:
"""
Converts the given Table into a DataStream of a specified type. The Table must only have
insert (append) changes. If the Table is also modified by update or delete changes, the
conversion will fail.
The fields of the Table are mapped to DataStream as follows: Row and Tuple types: Fields are
mapped by position, field types must match.
:param table: The Table to convert.
:param type_info: The TypeInformation that specifies the type of the DataStream.
:return: The converted DataStream.
.. versionadded:: 1.12.0
"""
j_data_stream = self._j_tenv.toAppendStream(table._j_table, type_info.get_java_type_info())
return DataStream(j_data_stream=j_data_stream)
def to_retract_stream(self, table: Table, type_info: TypeInformation) -> DataStream:
"""
Converts the given Table into a DataStream of add and retract messages. The message will be
encoded as Tuple. The first field is a boolean flag, the second field holds the record of
the specified type.
A true flag indicates an add message, a false flag indicates a retract message.
The fields of the Table are mapped to DataStream as follows: Row and Tuple types: Fields are
mapped by position, field types must match.
:param table: The Table to convert.
:param type_info: The TypeInformation of the requested record type.
:return: The converted DataStream.
.. versionadded:: 1.12.0
"""
j_data_stream = self._j_tenv.toRetractStream(table._j_table, type_info.get_java_type_info())
return DataStream(j_data_stream=j_data_stream)