You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2020/05/08 11:42:16 UTC
[flink] branch master updated: [FLINK-17256][python] Support
keyword arguments in the PyFlink descriptor API
This is an automated email from the ASF dual-hosted git repository.
dianfu 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 0d1738b [FLINK-17256][python] Support keyword arguments in the PyFlink descriptor API
0d1738b is described below
commit 0d1738bb15c19fb6c4efe4e7fef55543c33edcea
Author: acqua.csq <ac...@alibaba-inc.com>
AuthorDate: Thu Apr 30 16:07:30 2020 +0800
[FLINK-17256][python] Support keyword arguments in the PyFlink descriptor API
This closes #12013.
---
flink-python/pyflink/table/descriptors.py | 398 ++++++++++++++++++++-
.../pyflink/table/tests/test_descriptor.py | 67 ++++
2 files changed, 454 insertions(+), 11 deletions(-)
diff --git a/flink-python/pyflink/table/descriptors.py b/flink-python/pyflink/table/descriptors.py
index 2329312..a51d0e5 100644
--- a/flink-python/pyflink/table/descriptors.py
+++ b/flink-python/pyflink/table/descriptors.py
@@ -15,7 +15,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import sys
from abc import ABCMeta
+from collections import OrderedDict
from py4j.java_gateway import get_method
@@ -181,11 +183,28 @@ class Schema(Descriptor):
Field names are matched by the exact name by default (case sensitive).
"""
- def __init__(self):
+ def __init__(self, schema=None, fields=None, rowtime=None):
+ """
+ Constructor of Schema descriptor.
+
+ :param schema: The :class:`TableSchema` object.
+ :param fields: Dict of fields with the field name and the data type or type string stored.
+ :param rowtime: A :class:`RowTime` that Specifies the previously defined field as an
+ event-time attribute.
+ """
gateway = get_gateway()
self._j_schema = gateway.jvm.Schema()
super(Schema, self).__init__(self._j_schema)
+ if schema is not None:
+ self.schema(schema)
+
+ if fields is not None:
+ self.fields(fields)
+
+ if rowtime is not None:
+ self.rowtime(rowtime)
+
def schema(self, table_schema):
"""
Sets the schema with field names and the types. Required.
@@ -216,6 +235,26 @@ class Schema(Descriptor):
self._j_schema = self._j_schema.field(field_name, _to_java_type(field_type))
return self
+ def fields(self, fields):
+ """
+ Adds a set of fields with the field name and the data type or type string stored in a
+ list.
+
+ :param fields: Dict of fields with the field name and the data type or type string
+ stored.
+ E.g, [('int_field', DataTypes.INT()), ('string_field', DataTypes.STRING())].
+ :return: This schema object.
+ """
+ if sys.version_info[:2] <= (3, 5) and not isinstance(fields, OrderedDict):
+ raise TypeError("Must use OrderedDict type in python3.5 or older version to key the "
+ "schema in insert order.")
+ elif sys.version_info[:2] > (3, 5) and not isinstance(fields, (OrderedDict, dict)):
+ raise TypeError("fields must be stored in a dict or OrderedDict")
+
+ for field_name, field_type in fields.items():
+ self.field(field_name=field_name, field_type=field_type)
+ return self
+
def from_origin_field(self, origin_field_name):
"""
Specifies the origin of the previously defined field. The origin field is defined by a
@@ -285,11 +324,45 @@ class OldCsv(FormatDescriptor):
Deprecated: use the RFC-compliant `Csv` format instead when writing to Kafka.
"""
- def __init__(self):
+ def __init__(self, schema=None, field_delimiter=None, line_delimiter=None, quote_character=None,
+ comment_prefix=None, ignore_parse_errors=False, ignore_first_line=False):
+ """
+ Constructor of OldCsv format descriptor.
+
+ :param schema: Data type from :class:`DataTypes` that describes the schema.
+ :param field_delimiter: The field delimiter character.
+ :param line_delimiter: The line delimiter.
+ :param quote_character: The quote character.
+ :param comment_prefix: The prefix to indicate comments.
+ :param ignore_parse_errors: Skip records with parse error instead to fail. Throw an
+ exception by default.
+ :param ignore_first_line: Ignore the first line. Not skip the first line by default.
+ """
gateway = get_gateway()
self._j_csv = gateway.jvm.OldCsv()
super(OldCsv, self).__init__(self._j_csv)
+ if schema is not None:
+ self.schema(schema)
+
+ if field_delimiter is not None:
+ self.field_delimiter(field_delimiter)
+
+ if line_delimiter is not None:
+ self.line_delimiter(line_delimiter)
+
+ if quote_character is not None:
+ self.quote_character(quote_character)
+
+ if comment_prefix is not None:
+ self.comment_prefix(comment_prefix)
+
+ if ignore_parse_errors:
+ self.ignore_parse_errors()
+
+ if ignore_first_line:
+ self.ignore_first_line()
+
def field_delimiter(self, delimiter):
"""
Sets the field delimiter, "," by default.
@@ -392,11 +465,55 @@ class Csv(FormatDescriptor):
still available as :class:`OldCsv` for stream/batch filesystem operations.
"""
- def __init__(self):
+ def __init__(self, schema=None, field_delimiter=None, line_delimiter=None, quote_character=None,
+ allow_comments=False, ignore_parse_errors=False, array_element_delimiter=None,
+ escape_character=None, null_literal=None):
+ """
+ Constructor of Csv format descriptor.
+
+ :param schema: Data type from :class:`DataTypes` that describes the schema.
+ :param field_delimiter: The field delimiter character.
+ :param line_delimiter: The line delimiter.
+ :param quote_character: The quote character.
+ :param allow_comments: Whether Ignores comment lines that start with '#' (disabled by
+ default).
+ :param ignore_parse_errors: Whether Skip records with parse error instead to fail. Throw an
+ exception by default.
+ :param array_element_delimiter: The array element delimiter.
+ :param escape_character: Escaping character (e.g. backslash).
+ :param null_literal: The null literal string.
+ """
gateway = get_gateway()
self._j_csv = gateway.jvm.Csv()
super(Csv, self).__init__(self._j_csv)
+ if schema is not None:
+ self.schema(schema)
+
+ if field_delimiter is not None:
+ self.field_delimiter(field_delimiter)
+
+ if line_delimiter is not None:
+ self.line_delimiter(line_delimiter)
+
+ if quote_character is not None:
+ self.quote_character(quote_character)
+
+ if allow_comments:
+ self.allow_comments()
+
+ if ignore_parse_errors:
+ self.ignore_parse_errors()
+
+ if array_element_delimiter is not None:
+ self.array_element_delimiter(array_element_delimiter)
+
+ if escape_character is not None:
+ self.escape_character(escape_character)
+
+ if null_literal is not None:
+ self.null_literal(null_literal)
+
def field_delimiter(self, delimiter):
"""
Sets the field delimiter character (',' by default).
@@ -477,7 +594,7 @@ class Csv(FormatDescriptor):
"""
Sets the null literal string that is interpreted as a null value (disabled by default).
- :param null_literal:
+ :param null_literal: The null literal string.
:return: This :class:`Csv` object.
"""
self._j_csv = self._j_csv.nullLiteral(null_literal)
@@ -514,11 +631,23 @@ class Avro(FormatDescriptor):
Format descriptor for Apache Avro records.
"""
- def __init__(self):
+ def __init__(self, record_class=None, avro_schema=None):
+ """
+ Constructor of Avro format descriptor.
+
+ :param record_class: The java fully-qualified class name of the Avro record.
+ :param avro_schema: Avro schema string.
+ """
gateway = get_gateway()
self._j_avro = gateway.jvm.Avro()
super(Avro, self).__init__(self._j_avro)
+ if record_class is not None:
+ self.record_class(record_class)
+
+ if avro_schema is not None:
+ self.avro_schema(avro_schema)
+
def record_class(self, record_class):
"""
Sets the class of the Avro specific record.
@@ -547,11 +676,27 @@ class Json(FormatDescriptor):
Format descriptor for JSON.
"""
- def __init__(self):
+ def __init__(self, json_schema=None, schema=None, derive_schema=False):
+ """
+ Constructor of Json format descriptor.
+
+ :param json_schema: The JSON schema string.
+ :param schema: Sets the schema using :class:`DataTypes` that describes the schema.
+ :param derive_schema: Derives the format schema from the table's schema described.
+ """
gateway = get_gateway()
self._j_json = gateway.jvm.Json()
super(Json, self).__init__(self._j_json)
+ if json_schema is not None:
+ self.json_schema(json_schema)
+
+ if schema is not None:
+ self.schema(schema)
+
+ if derive_schema:
+ self.derive_schema()
+
def fail_on_missing_field(self, fail_on_missing_field):
"""
Sets flag whether to fail if a field is missing or not.
@@ -691,11 +836,19 @@ class FileSystem(ConnectorDescriptor):
Connector descriptor for a file system.
"""
- def __init__(self):
+ def __init__(self, path=None):
+ """
+ Constructor of FileSystem descriptor.
+
+ :param path: The path of a file or directory.
+ """
gateway = get_gateway()
self._j_file_system = gateway.jvm.FileSystem()
super(FileSystem, self).__init__(self._j_file_system)
+ if path is not None:
+ self.path(path)
+
def path(self, path_str):
"""
Sets the path to a file or directory in a file system.
@@ -712,11 +865,72 @@ class Kafka(ConnectorDescriptor):
Connector descriptor for the Apache Kafka message queue.
"""
- def __init__(self):
+ def __init__(self, version=None, topic=None, properties=None, start_from_earliest=False,
+ start_from_latest=False, start_from_group_offsets=True,
+ start_from_specific_offsets_dict=None, start_from_timestamp=None,
+ sink_partitioner_fixed=None, sink_partitioner_round_robin=None,
+ custom_partitioner_class_name=None):
+ """
+ Constructor of Kafka descriptor.
+
+ :param version: Kafka version. E.g., "0.8", "0.11", etc.
+ :param topic: The topic from which the table is read.
+ :param properties: The dict object contains configuration properties for the Kafka
+ consumer. Both the keys and values should be strings.
+ :param start_from_earliest: Specifies the consumer to start reading from the earliest offset
+ for all partitions.
+ :param start_from_latest: Specifies the consumer to start reading from the latest offset for
+ all partitions.
+ :param start_from_group_offsets: Specifies the consumer to start reading from any committed
+ group offsets found in Zookeeper / Kafka brokers.
+ :param start_from_specific_offsets_dict: Dict of specific_offsets that the key is int-type
+ partition id and value is int-type offset value.
+ :param start_from_timestamp: Specifies the consumer to start reading partitions from a
+ specified timestamp.
+ :param sink_partitioner_fixed: Configures how to partition records from Flink's partitions
+ into Kafka's partitions.
+ :param sink_partitioner_round_robin: Configures how to partition records from Flink's
+ partitions into Kafka's partitions.
+ :param custom_partitioner_class_name: Configures how to partition records from Flink's
+ partitions into Kafka's partitions.
+ """
gateway = get_gateway()
self._j_kafka = gateway.jvm.Kafka()
super(Kafka, self).__init__(self._j_kafka)
+ if version is not None:
+ self.version(version)
+
+ if topic is not None:
+ self.topic(topic)
+
+ if properties is not None:
+ self.properties(properties)
+
+ if start_from_earliest:
+ self.start_from_earliest()
+
+ if start_from_latest:
+ self.start_from_latest()
+
+ if start_from_group_offsets:
+ self.start_from_group_offsets()
+
+ if start_from_specific_offsets_dict is not None:
+ self.start_from_specific_offsets(start_from_specific_offsets_dict)
+
+ if start_from_timestamp is not None:
+ self.start_from_timestamp(start_from_timestamp)
+
+ if sink_partitioner_fixed is not None and sink_partitioner_fixed is True:
+ self.sink_partitioner_fixed()
+
+ if sink_partitioner_round_robin is not None and sink_partitioner_round_robin is True:
+ self.sink_partitioner_round_robin()
+
+ if custom_partitioner_class_name is not None:
+ self.sink_partitioner_custom(custom_partitioner_class_name)
+
def version(self, version):
"""
Sets the Kafka version to be used.
@@ -948,11 +1162,113 @@ class Elasticsearch(ConnectorDescriptor):
Connector descriptor for the Elasticsearch search engine.
"""
- def __init__(self):
+ def __init__(self, version=None, hostname=None, port=None, protocol=None, index=None,
+ document_type=None, key_delimiter=None, key_null_literal=None,
+ failure_handler_fail=False, failure_handler_ignore=False,
+ failure_handler_retry_rejected=False, custom_failure_handler_class_name=None,
+ disable_flush_on_checkpoint=False, bulk_flush_max_actions=None,
+ bulk_flush_max_size=None, bulk_flush_interval=None,
+ bulk_flush_backoff_constant=False, bulk_flush_backoff_exponential=False,
+ bulk_flush_backoff_max_retries=None, bulk_flush_backoff_delay=None,
+ connection_max_retry_timeout=None, connection_path_prefix=None):
+ """
+ Constructor of Elasticsearch descriptor.
+
+ :param version: Elasticsearch version. E.g., "6".
+ :param hostname: Connection hostname.
+ :param port: Connection port.
+ :param protocol: Connection protocol; e.g. "http".
+ :param index: Elasticsearch index.
+ :param document_type: Elasticsearch document type.
+ :param key_delimiter: Key delimiter; e.g., "$" would result in IDs "KEY1$KEY2$KEY3".
+ :param key_null_literal: key null literal string; e.g. "N/A" would result in IDs
+ "KEY1_N/A_KEY3".
+ :param failure_handler_fail: Configures a failure handling strategy in case a request to
+ Elasticsearch fails.
+ :param failure_handler_ignore: Configures a failure handling strategy in case a request to
+ Elasticsearch fails.
+ :param failure_handler_retry_rejected: Configures a failure handling strategy in case a
+ request to Elasticsearch fails.
+ :param custom_failure_handler_class_name: Configures a failure handling strategy in case a
+ request to Elasticsearch fails.
+ :param disable_flush_on_checkpoint: Disables flushing on checkpoint
+ :param bulk_flush_max_actions: the maximum number of actions to buffer per bulk request.
+ :param bulk_flush_max_size: The maximum size. E.g. "42 mb". only MB granularity is
+ supported.
+ :param bulk_flush_interval: Bulk flush interval (in milliseconds).
+ :param bulk_flush_backoff_constant: Configures how to buffer elements before sending them in
+ bulk to the cluster for efficiency.
+ :param bulk_flush_backoff_exponential: Configures how to buffer elements before sending them
+ in bulk to the cluster for efficiency.
+ :param bulk_flush_backoff_max_retries: The maximum number of retries.
+ :param bulk_flush_backoff_delay: Delay between each backoff attempt (in milliseconds).
+ :param connection_max_retry_timeout: Maximum timeout (in milliseconds).
+ :param connection_path_prefix: Prefix string to be added to every REST communication.
+ """
gateway = get_gateway()
self._j_elasticsearch = gateway.jvm.Elasticsearch()
super(Elasticsearch, self).__init__(self._j_elasticsearch)
+ if version is not None:
+ self.version(version)
+
+ if hostname is not None:
+ self.host(hostname, port, protocol)
+
+ if index is not None:
+ self.index(index)
+
+ if document_type is not None:
+ self.document_type(document_type)
+
+ if key_delimiter is not None:
+ self.key_delimiter(key_delimiter)
+
+ if key_null_literal is not None:
+ self.key_null_literal(key_null_literal)
+
+ if failure_handler_fail:
+ self.failure_handler_fail()
+
+ if failure_handler_ignore:
+ self.failure_handler_ignore()
+
+ if failure_handler_retry_rejected:
+ self.failure_handler_retry_rejected()
+
+ if custom_failure_handler_class_name is not None:
+ self.failure_handler_custom(custom_failure_handler_class_name)
+
+ if disable_flush_on_checkpoint:
+ self.disable_flush_on_checkpoint()
+
+ if bulk_flush_max_actions is not None:
+ self.bulk_flush_max_actions(bulk_flush_max_actions)
+
+ if bulk_flush_max_size is not None:
+ self.bulk_flush_max_size(bulk_flush_max_size)
+
+ if bulk_flush_interval is not None:
+ self.bulk_flush_interval(bulk_flush_interval)
+
+ if bulk_flush_backoff_constant:
+ self.bulk_flush_backoff_constant()
+
+ if bulk_flush_backoff_exponential:
+ self.bulk_flush_backoff_exponential()
+
+ if bulk_flush_backoff_max_retries is not None:
+ self.bulk_flush_backoff_max_retries(bulk_flush_backoff_max_retries)
+
+ if bulk_flush_backoff_delay is not None:
+ self.bulk_flush_backoff_delay(bulk_flush_backoff_delay)
+
+ if connection_max_retry_timeout is not None:
+ self.connection_max_retry_timeout(connection_max_retry_timeout)
+
+ if connection_path_prefix is not None:
+ self.connection_path_prefix(connection_path_prefix)
+
def version(self, version):
"""
Sets the Elasticsearch version to be used. Required.
@@ -1214,11 +1530,50 @@ class HBase(ConnectorDescriptor):
Connector descriptor for Apache HBase.
"""
- def __init__(self):
+ def __init__(self, version=None, table_name=None, zookeeper_quorum=None,
+ zookeeper_node_parent=None, write_buffer_flush_max_size=None,
+ write_buffer_flush_max_rows=None, write_buffer_flush_interval=None):
+ """
+ Constructor of HBase descriptor.
+
+ :param version: HBase version. E.g., "1.4.3".
+ :param table_name: Name of HBase table. E.g., "testNamespace:testTable", "testDefaultTable"
+ :param zookeeper_quorum: zookeeper quorum address to connect the HBase cluster. E.g.,
+ "localhost:2181,localhost:2182,localhost:2183"
+ :param zookeeper_node_parent: zookeeper node path of hbase cluster. E.g,
+ "/hbase/example-root-znode".
+ :param write_buffer_flush_max_size: the maximum size.
+ :param write_buffer_flush_max_rows: number of added rows when begin the request flushing.
+ :param write_buffer_flush_interval: flush interval. The string should be in format
+ "{length value}{time unit label}" E.g, "123ms", "1 s",
+ if not time unit label is specified, it will be
+ considered as milliseconds.
+ """
gateway = get_gateway()
self._j_hbase = gateway.jvm.HBase()
super(HBase, self).__init__(self._j_hbase)
+ if version is not None:
+ self.version(version)
+
+ if table_name is not None:
+ self.table_name(table_name)
+
+ if zookeeper_quorum is not None:
+ self.zookeeper_quorum(zookeeper_quorum)
+
+ if zookeeper_node_parent is not None:
+ self.zookeeper_node_parent(zookeeper_node_parent)
+
+ if write_buffer_flush_max_size is not None:
+ self.write_buffer_flush_max_size(write_buffer_flush_max_size)
+
+ if write_buffer_flush_max_rows is not None:
+ self.write_buffer_flush_max_rows(write_buffer_flush_max_rows)
+
+ if write_buffer_flush_interval is not None:
+ self.write_buffer_flush_interval(write_buffer_flush_interval)
+
def version(self, version):
"""
Set the Apache HBase version to be used, Required.
@@ -1420,10 +1775,31 @@ class StreamTableDescriptor(ConnectTableDescriptor):
.. seealso:: parent class: :class:`ConnectTableDescriptor`
"""
- def __init__(self, j_stream_table_descriptor):
+ def __init__(self, j_stream_table_descriptor, in_append_mode=False, in_retract_mode=False,
+ in_upsert_mode=False):
+ """
+ Constructor of StreamTableDescriptor.
+
+ :param j_stream_table_descriptor: the corresponding stream table descriptor in java.
+ :param in_append_mode: Declares the conversion between a dynamic table and external
+ connector in append mode.
+ :param in_retract_mode: Declares the conversion between a dynamic table and external
+ connector in retract mode.
+ :param in_upsert_mode: Declares the conversion between a dynamic table and external
+ connector in upsert mode.
+ """
self._j_stream_table_descriptor = j_stream_table_descriptor
super(StreamTableDescriptor, self).__init__(self._j_stream_table_descriptor)
+ if in_append_mode:
+ self.in_append_mode()
+
+ if in_retract_mode:
+ self.in_retract_mode()
+
+ if in_upsert_mode:
+ self.in_upsert_mode()
+
def in_append_mode(self):
"""
Declares how to perform the conversion between a dynamic table and an external connector.
diff --git a/flink-python/pyflink/table/tests/test_descriptor.py b/flink-python/pyflink/table/tests/test_descriptor.py
index 1b85140..beb8d8a 100644
--- a/flink-python/pyflink/table/tests/test_descriptor.py
+++ b/flink-python/pyflink/table/tests/test_descriptor.py
@@ -15,7 +15,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import collections
import os
+import sys
from pyflink.table.descriptors import (FileSystem, OldCsv, Rowtime, Schema, Kafka,
Elasticsearch, Csv, Avro, Json, CustomConnectorDescriptor,
@@ -48,6 +50,7 @@ class KafkaDescriptorTests(PyFlinkTestCase):
properties = kafka.to_properties()
expected = {'connector.version': '0.11',
'connector.type': 'kafka',
+ 'connector.startup-mode': 'group-offsets',
'connector.property-version': '1'}
self.assertEqual(expected, properties)
@@ -57,6 +60,7 @@ class KafkaDescriptorTests(PyFlinkTestCase):
properties = kafka.to_properties()
expected = {'connector.type': 'kafka',
'connector.topic': 'topic1',
+ 'connector.startup-mode': 'group-offsets',
'connector.property-version': '1'}
self.assertEqual(expected, properties)
@@ -65,6 +69,7 @@ class KafkaDescriptorTests(PyFlinkTestCase):
properties = kafka.to_properties()
expected = {'connector.type': 'kafka',
+ 'connector.startup-mode': 'group-offsets',
'connector.properties.bootstrap.servers': 'localhost:9092',
'connector.property-version': '1'}
self.assertEqual(expected, properties)
@@ -74,6 +79,7 @@ class KafkaDescriptorTests(PyFlinkTestCase):
properties = kafka.to_properties()
expected = {'connector.type': 'kafka',
+ 'connector.startup-mode': 'group-offsets',
'connector.properties.group.id': 'testGroup',
'connector.property-version': '1'}
self.assertEqual(expected, properties)
@@ -130,6 +136,7 @@ class KafkaDescriptorTests(PyFlinkTestCase):
properties = kafka.to_properties()
expected = {'connector.sink-partitioner': 'fixed',
+ 'connector.startup-mode': 'group-offsets',
'connector.type': 'kafka',
'connector.property-version': '1'}
self.assertEqual(expected, properties)
@@ -144,6 +151,7 @@ class KafkaDescriptorTests(PyFlinkTestCase):
'org.apache.flink.streaming.connectors.kafka.partitioner.'
'FlinkFixedPartitioner',
'connector.type': 'kafka',
+ 'connector.startup-mode': 'group-offsets',
'connector.property-version': '1'}
self.assertEqual(expected, properties)
@@ -153,6 +161,7 @@ class KafkaDescriptorTests(PyFlinkTestCase):
properties = kafka.to_properties()
expected = {'connector.sink-partitioner': 'round-robin',
'connector.type': 'kafka',
+ 'connector.startup-mode': 'group-offsets',
'connector.property-version': '1'}
self.assertEqual(expected, properties)
@@ -905,6 +914,64 @@ class SchemaDescriptorTests(PyFlinkTestCase):
'schema.10.data-type': 'BOOLEAN'}
self.assertEqual(expected, properties)
+ def test_fields(self):
+ fields = collections.OrderedDict([
+ ("int_field", DataTypes.INT()),
+ ("long_field", DataTypes.BIGINT()),
+ ("string_field", DataTypes.STRING()),
+ ("timestamp_field", DataTypes.TIMESTAMP(3)),
+ ("time_field", DataTypes.TIME()),
+ ("date_field", DataTypes.DATE()),
+ ("double_field", DataTypes.DOUBLE()),
+ ("float_field", DataTypes.FLOAT()),
+ ("byte_field", DataTypes.TINYINT()),
+ ("short_field", DataTypes.SMALLINT()),
+ ("boolean_field", DataTypes.BOOLEAN())
+ ])
+
+ schema = Schema().fields(fields)
+
+ properties = schema.to_properties()
+ expected = {'schema.0.name': 'int_field',
+ 'schema.0.data-type': 'INT',
+ 'schema.1.name': 'long_field',
+ 'schema.1.data-type': 'BIGINT',
+ 'schema.2.name': 'string_field',
+ 'schema.2.data-type': 'VARCHAR(2147483647)',
+ 'schema.3.name': 'timestamp_field',
+ 'schema.3.data-type': 'TIMESTAMP(3)',
+ 'schema.4.name': 'time_field',
+ 'schema.4.data-type': 'TIME(0)',
+ 'schema.5.name': 'date_field',
+ 'schema.5.data-type': 'DATE',
+ 'schema.6.name': 'double_field',
+ 'schema.6.data-type': 'DOUBLE',
+ 'schema.7.name': 'float_field',
+ 'schema.7.data-type': 'FLOAT',
+ 'schema.8.name': 'byte_field',
+ 'schema.8.data-type': 'TINYINT',
+ 'schema.9.name': 'short_field',
+ 'schema.9.data-type': 'SMALLINT',
+ 'schema.10.name': 'boolean_field',
+ 'schema.10.data-type': 'BOOLEAN'}
+ self.assertEqual(expected, properties)
+
+ if sys.version_info[:2] <= (3, 5):
+ fields = {
+ "int_field": DataTypes.INT(),
+ "long_field": DataTypes.BIGINT(),
+ "string_field": DataTypes.STRING(),
+ "timestamp_field": DataTypes.TIMESTAMP(3),
+ "time_field": DataTypes.TIME(),
+ "date_field": DataTypes.DATE(),
+ "double_field": DataTypes.DOUBLE(),
+ "float_field": DataTypes.FLOAT(),
+ "byte_field": DataTypes.TINYINT(),
+ "short_field": DataTypes.SMALLINT(),
+ "boolean_field": DataTypes.BOOLEAN()
+ }
+ self.assertRaises(TypeError, Schema().fields, fields)
+
def test_field_in_string(self):
schema = Schema()\
.field("int_field", 'INT')\