You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by he...@apache.org on 2020/08/18 05:13:47 UTC
[flink] branch master updated: [FLINK-18944][python] Support JDBC
connector for Python DataStream API. (#13169)
This is an automated email from the ASF dual-hosted git repository.
hequn 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 f674c7c [FLINK-18944][python] Support JDBC connector for Python DataStream API. (#13169)
f674c7c is described below
commit f674c7c9c5bf733b77e661e9ecc74c60d7fc906e
Author: Shuiqiang Chen <ac...@alibaba-inc.com>
AuthorDate: Tue Aug 18 13:11:59 2020 +0800
[FLINK-18944][python] Support JDBC connector for Python DataStream API. (#13169)
---
flink-python/pyflink/datastream/connectors.py | 146 +++++++++++++++++++++
flink-python/pyflink/datastream/data_stream.py | 30 +++++
.../pyflink/datastream/tests/test_connectors.py | 52 +++++++-
3 files changed, 227 insertions(+), 1 deletion(-)
diff --git a/flink-python/pyflink/datastream/connectors.py b/flink-python/pyflink/datastream/connectors.py
index 7c3dfa0..a89ef7b 100644
--- a/flink-python/pyflink/datastream/connectors.py
+++ b/flink-python/pyflink/datastream/connectors.py
@@ -20,8 +20,10 @@ from typing import Dict, List, Union
from pyflink.common import typeinfo
from pyflink.common.serialization_schemas import DeserializationSchema, SerializationSchema
+from pyflink.common.typeinfo import RowTypeInfo
from pyflink.datastream.functions import SourceFunction, SinkFunction
from pyflink.java_gateway import get_gateway
+from pyflink.util.utils import to_jarray
class FlinkKafkaConsumerBase(SourceFunction, abc.ABC):
@@ -487,3 +489,147 @@ def _get_kafka_consumer(topics, properties, deserialization_schema, j_consumer_c
deserialization_schema._j_deserialization_schema,
j_properties)
return j_flink_kafka_consumer
+
+
+class JdbcSink(SinkFunction):
+
+ def __init__(self, j_jdbc_sink):
+ super(JdbcSink, self).__init__(sink_func=j_jdbc_sink)
+
+ @staticmethod
+ def sink(sql: str, type_info: RowTypeInfo, jdbc_connection_options: 'JdbcConnectionOptions',
+ jdbc_execution_options: 'JdbcExecutionOptions' = None):
+ """
+ Create a JDBC sink.
+
+ :param sql: arbitrary DML query (e.g. insert, update, upsert)
+ :param type_info: A RowTypeInfo for query field types.
+ :param jdbc_execution_options: parameters of execution, such as batch size and maximum
+ retries.
+ :param jdbc_connection_options: parameters of connection, such as JDBC URL.
+ :return: A JdbcSink.
+ """
+ sql_types = []
+ gateway = get_gateway()
+ JJdbcTypeUtil = gateway.jvm.org.apache.flink.connector.jdbc.utils.JdbcTypeUtil
+ for field_type in type_info.get_field_types():
+ sql_types.append(JJdbcTypeUtil
+ .typeInformationToSqlType(field_type.get_java_type_info()))
+ j_sql_type = to_jarray(gateway.jvm.int, sql_types)
+ output_format_clz = gateway.jvm.Class\
+ .forName('org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat', False,
+ get_gateway().jvm.Thread.currentThread().getContextClassLoader())
+ j_int_array_type = to_jarray(gateway.jvm.int, []).getClass()
+ j_builder_method = output_format_clz.getDeclaredMethod('createRowJdbcStatementBuilder',
+ to_jarray(gateway.jvm.Class,
+ [j_int_array_type]))
+ j_builder_method.setAccessible(True)
+ j_statement_builder = j_builder_method.invoke(None, to_jarray(gateway.jvm.Object,
+ [j_sql_type]))
+
+ jdbc_execution_options = jdbc_execution_options if jdbc_execution_options is not None \
+ else JdbcExecutionOptions.defaults()
+ j_jdbc_sink = gateway.jvm.org.apache.flink.connector.jdbc.JdbcSink\
+ .sink(sql, j_statement_builder, jdbc_execution_options._j_jdbc_execution_options,
+ jdbc_connection_options._j_jdbc_connection_options)
+ return JdbcSink(j_jdbc_sink=j_jdbc_sink)
+
+
+class JdbcConnectionOptions(object):
+ """
+ JDBC connection options.
+ """
+ def __init__(self, j_jdbc_connection_options):
+ self._j_jdbc_connection_options = j_jdbc_connection_options
+
+ def get_db_url(self) -> str:
+ return self._j_jdbc_connection_options.getDbURL()
+
+ def get_driver_name(self) -> str:
+ return self._j_jdbc_connection_options.getDriverName()
+
+ def get_user_name(self) -> str:
+ return self._j_jdbc_connection_options.getUsername()
+
+ def get_password(self) -> str:
+ return self._j_jdbc_connection_options.getPassword()
+
+ class JdbcConnectionOptionsBuilder(object):
+ """
+ Builder for JdbcConnectionOptions.
+ """
+ def __init__(self):
+ self._j_options_builder = get_gateway().jvm.org.apache.flink.connector\
+ .jdbc.JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
+
+ def with_url(self, url: str) -> 'JdbcConnectionOptions.JdbcConnectionOptionsBuilder':
+ self._j_options_builder.withUrl(url)
+ return self
+
+ def with_driver_name(self, driver_name: str) \
+ -> 'JdbcConnectionOptions.JdbcConnectionOptionsBuilder':
+ self._j_options_builder.withDriverName(driver_name)
+ return self
+
+ def with_user_name(self, user_name: str) \
+ -> 'JdbcConnectionOptions.JdbcConnectionOptionsBuilder':
+ self._j_options_builder.withUsername(user_name)
+ return self
+
+ def with_password(self, password: str) \
+ -> 'JdbcConnectionOptions.JdbcConnectionOptionsBuilder':
+ self._j_options_builder.withPassword(password)
+ return self
+
+ def build(self) -> 'JdbcConnectionOptions':
+ return JdbcConnectionOptions(j_jdbc_connection_options=self._j_options_builder.build())
+
+
+class JdbcExecutionOptions(object):
+ """
+ JDBC sink batch options.
+ """
+ def __init__(self, j_jdbc_execution_options):
+ self._j_jdbc_execution_options = j_jdbc_execution_options
+
+ def get_batch_interval_ms(self) -> int:
+ return self._j_jdbc_execution_options.getBatchIntervalMs()
+
+ def get_batch_size(self) -> int:
+ return self._j_jdbc_execution_options.getBatchSize()
+
+ def get_max_retries(self) -> int:
+ return self._j_jdbc_execution_options.getMaxRetries()
+
+ @staticmethod
+ def defaults() -> 'JdbcExecutionOptions':
+ return JdbcExecutionOptions(
+ j_jdbc_execution_options=get_gateway().jvm
+ .org.apache.flink.connector.jdbc.JdbcExecutionOptions.defaults())
+
+ @staticmethod
+ def builder() -> 'Builder':
+ return JdbcExecutionOptions.Builder()
+
+ class Builder(object):
+ """
+ Builder for JdbcExecutionOptions.
+ """
+ def __init__(self):
+ self._j_builder = get_gateway().jvm\
+ .org.apache.flink.connector.jdbc.JdbcExecutionOptions.builder()
+
+ def with_batch_size(self, size: int) -> 'JdbcExecutionOptions.Builder':
+ self._j_builder.withBatchSize(size)
+ return self
+
+ def with_batch_interval_ms(self, interval_ms: int) -> 'JdbcExecutionOptions.Builder':
+ self._j_builder.withBatchIntervalMs(interval_ms)
+ return self
+
+ def with_max_retries(self, max_retries: int) -> 'JdbcExecutionOptions.Builder':
+ self._j_builder.withMaxRetries(max_retries)
+ return self
+
+ def build(self) -> 'JdbcExecutionOptions':
+ return JdbcExecutionOptions(j_jdbc_execution_options=self._j_builder.build())
diff --git a/flink-python/pyflink/datastream/data_stream.py b/flink-python/pyflink/datastream/data_stream.py
index 2c5f526..80f1962 100644
--- a/flink-python/pyflink/datastream/data_stream.py
+++ b/flink-python/pyflink/datastream/data_stream.py
@@ -712,6 +712,36 @@ class DataStreamSink(object):
self._j_data_stream_sink.setParallelism(parallelism)
return self
+ def disable_chaining(self) -> 'DataStreamSink':
+ """
+ Turns off chaining for this operator so thread co-location will not be used as an
+ optimization.
+ Chaining can be turned off for the whole job by
+ StreamExecutionEnvironment.disableOperatorChaining() however it is not advised for
+ performance consideration.
+
+ :return: The operator with chaining disabled.
+ """
+ self._j_data_stream_sink.disableChaining()
+ return self
+
+ def slot_sharing_group(self, slot_sharing_group: str) -> 'DataStreamSink':
+ """
+ Sets the slot sharing group of this operation. Parallel instances of operations that are in
+ the same slot sharing group will be co-located in the same TaskManager slot, if possible.
+
+ Operations inherit the slot sharing group of input operations if all input operations are in
+ the same slot sharing group and no slot sharing group was explicitly specified.
+
+ Initially an operation is in the default slot sharing group. An operation can be put into
+ the default group explicitly by setting the slot sharing group to 'default'.
+
+ :param slot_sharing_group: The slot sharing group name.
+ :return: This operator.
+ """
+ self._j_data_stream_sink.slotSharingGroup(slot_sharing_group)
+ return self
+
class KeyedStream(DataStream):
"""
diff --git a/flink-python/pyflink/datastream/tests/test_connectors.py b/flink-python/pyflink/datastream/tests/test_connectors.py
index c179f72..1052e87 100644
--- a/flink-python/pyflink/datastream/tests/test_connectors.py
+++ b/flink-python/pyflink/datastream/tests/test_connectors.py
@@ -22,7 +22,8 @@ from pyflink.common.serialization_schemas import JsonRowDeserializationSchema, \
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer010, FlinkKafkaProducer010, \
- FlinkKafkaConsumer011, FlinkKafkaProducer011, FlinkKafkaConsumer, FlinkKafkaProducer
+ FlinkKafkaConsumer011, FlinkKafkaProducer011, FlinkKafkaConsumer, FlinkKafkaProducer, JdbcSink,\
+ JdbcConnectionOptions, JdbcExecutionOptions
from pyflink.java_gateway import get_gateway
from pyflink.testing.test_case_utils import PyFlinkTestCase, _load_specific_flink_module_jars, \
get_private_field, invoke_java_object_method
@@ -100,3 +101,52 @@ class FlinkKafkaTest(PyFlinkTestCase):
# Change the ClassLoader back to the cached ContextClassLoader after the test case finished.
if self._cxt_clz_loader is not None:
get_gateway().jvm.Thread.currentThread().setContextClassLoader(self._cxt_clz_loader)
+
+
+class FlinkJdbcSinkTest(PyFlinkTestCase):
+
+ def setUp(self) -> None:
+ self.env = StreamExecutionEnvironment.get_execution_environment()
+ self._cxt_clz_loader = get_gateway().jvm.Thread.currentThread().getContextClassLoader()
+ _load_specific_flink_module_jars('/flink-connectors/flink-connector-jdbc')
+
+ def test_jdbc_sink(self):
+ ds = self.env.from_collection([('ab', 1), ('bdc', 2), ('cfgs', 3), ('deeefg', 4)],
+ type_info=Types.ROW([Types.STRING(), Types.INT()]))
+ jdbc_connection_options = JdbcConnectionOptions.JdbcConnectionOptionsBuilder()\
+ .with_driver_name('com.mysql.jdbc.Driver')\
+ .with_user_name('root')\
+ .with_password('password')\
+ .with_url('jdbc:mysql://server-name:server-port/database-name').build()
+
+ jdbc_execution_options = JdbcExecutionOptions.builder().with_batch_interval_ms(2000)\
+ .with_batch_size(100).with_max_retries(5).build()
+ jdbc_sink = JdbcSink.sink("insert into test table", ds.get_type(), jdbc_connection_options,
+ jdbc_execution_options)
+
+ ds.add_sink(jdbc_sink).name('jdbc sink')
+ plan = eval(self.env.get_execution_plan())
+ self.assertEqual('Sink: jdbc sink', plan['nodes'][1]['type'])
+ j_output_format = get_private_field(jdbc_sink.get_java_function(), 'outputFormat')
+
+ connection_options = JdbcConnectionOptions(
+ get_private_field(get_private_field(j_output_format, 'connectionProvider'),
+ 'jdbcOptions'))
+ self.assertEqual(jdbc_connection_options.get_db_url(), connection_options.get_db_url())
+ self.assertEqual(jdbc_connection_options.get_driver_name(),
+ connection_options.get_driver_name())
+ self.assertEqual(jdbc_connection_options.get_password(), connection_options.get_password())
+ self.assertEqual(jdbc_connection_options.get_user_name(),
+ connection_options.get_user_name())
+
+ exec_options = JdbcExecutionOptions(get_private_field(j_output_format, 'executionOptions'))
+ self.assertEqual(jdbc_execution_options.get_batch_interval_ms(),
+ exec_options.get_batch_interval_ms())
+ self.assertEqual(jdbc_execution_options.get_batch_size(),
+ exec_options.get_batch_size())
+ self.assertEqual(jdbc_execution_options.get_max_retries(),
+ exec_options.get_max_retries())
+
+ def tearDown(self):
+ if self._cxt_clz_loader is not None:
+ get_gateway().jvm.Thread.currentThread().setContextClassLoader(self._cxt_clz_loader)