You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2019/12/24 02:58:00 UTC
[flink] 01/02: [FLINK-15193][table][docs] Move DDL to the first tab
in table connector page
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b61ae9eeba8fa163dbf02426fbffc685fa7ed000
Author: JingsongLi <lz...@aliyun.com>
AuthorDate: Mon Dec 23 16:08:07 2019 +0800
[FLINK-15193][table][docs] Move DDL to the first tab in table connector page
---
docs/dev/table/connect.md | 750 ++++++++++++++++++++--------------------
docs/dev/table/connect.zh.md | 806 +++++++++++++++++++++----------------------
2 files changed, 778 insertions(+), 778 deletions(-)
diff --git a/docs/dev/table/connect.md b/docs/dev/table/connect.md
index 40861f3..f4c8e39 100644
--- a/docs/dev/table/connect.md
+++ b/docs/dev/table/connect.md
@@ -92,6 +92,21 @@ The **table schema** defines the schema of a table that is exposed to SQL querie
The subsequent sections will cover each definition part ([connector](connect.html#table-connectors), [format](connect.html#table-formats), and [schema](connect.html#table-schema)) in more detail. The following example shows how to pass them:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+tableEnvironment.sqlUpdate(
+ "CREATE TABLE MyTable (\n" +
+ " ... -- declare table schema \n" +
+ ") WITH (\n" +
+ " 'connector.type' = '...', -- declare connector specific properties\n" +
+ " ...\n" +
+ " 'update-mode' = 'append', -- declare update mode\n" +
+ " 'format.type' = '...', -- declare format specific properties\n" +
+ " ...\n" +
+ ")");
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
tableEnvironment
@@ -124,21 +139,6 @@ format: ...
schema: ...
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-tableEnvironment.sqlUpdate(
- "CREATE TABLE MyTable (\n" +
- " ... -- declare table schema \n" +
- ") WITH (\n" +
- " 'connector.type' = '...', -- declare connector specific properties\n" +
- " ...\n" +
- " 'update-mode' = 'append', -- declare update mode\n" +
- " 'format.type' = '...', -- declare format specific properties\n" +
- " ...\n" +
- ")");
-{% endhighlight %}
-</div>
</div>
The table's type (`source`, `sink`, or `both`) determines how a table is registered. In case of table type `both`, both a table source and table sink are registered under the same name. Logically, this means that we can both read and write to such a table similarly to a table in a regular DBMS.
@@ -148,6 +148,41 @@ For streaming queries, an [update mode](connect.html#update-mode) declares how t
The following code shows a full example of how to connect to Kafka for reading Avro records.
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ -- declare the schema of the table
+ `user` BIGINT,
+ message STRING,
+ ts STRING
+) WITH (
+ -- declare the external system to connect to
+ 'connector.type' = 'kafka',
+ 'connector.version' = '0.10',
+ 'connector.topic' = 'topic_name',
+ 'connector.startup-mode' = 'earliest-offset',
+ 'connector.properties.zookeeper.connect' = 'localhost:2181',
+ 'connector.properties.bootstrap.servers' = 'localhost:9092',
+
+ -- specify the update-mode for streaming tables
+ 'update-mode' = 'append',
+
+ -- declare a format for this system
+ 'format.type' = 'avro',
+ 'format.avro-schema' = '{
+ "namespace": "org.myorganization",
+ "type": "record",
+ "name": "UserMessage",
+ "fields": [
+ {"name": "ts", "type": "string"},
+ {"name": "user", "type": "long"},
+ {"name": "message", "type": ["string", "null"]}
+ ]
+ }'
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
tableEnvironment
@@ -285,41 +320,6 @@ tables:
data-type: STRING
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- -- declare the schema of the table
- `user` BIGINT,
- message STRING,
- ts STRING
-) WITH (
- -- declare the external system to connect to
- 'connector.type' = 'kafka',
- 'connector.version' = '0.10',
- 'connector.topic' = 'topic_name',
- 'connector.startup-mode' = 'earliest-offset',
- 'connector.properties.zookeeper.connect' = 'localhost:2181',
- 'connector.properties.bootstrap.servers' = 'localhost:9092',
-
- -- specify the update-mode for streaming tables
- 'update-mode' = 'append',
-
- -- declare a format for this system
- 'format.type' = 'avro',
- 'format.avro-schema' = '{
- "namespace": "org.myorganization",
- "type": "record",
- "name": "UserMessage",
- "fields": [
- {"name": "ts", "type": "string"},
- {"name": "user", "type": "long"},
- {"name": "message", "type": ["string", "null"]}
- ]
- }'
-)
-{% endhighlight %}
-</div>
</div>
In both ways the desired connection properties are converted into normalized, string-based key-value pairs. So-called [table factories](sourceSinks.html#define-a-tablefactory) create configured table sources, table sinks, and corresponding formats from the key-value pairs. All table factories that can be found via Java's [Service Provider Interfaces (SPI)](https://docs.oracle.com/javase/tutorial/sound/SPI-intro.html) are taken into account when searching for exactly-one matching table factory.
@@ -600,6 +600,16 @@ For streaming queries, it is required to declare how to perform the [conversion
<span class="label label-danger">Attention</span> The documentation of each connector states which update modes are supported.
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyTable (
+ ...
+) WITH (
+ 'update-mode' = 'append' -- otherwise: 'retract' or 'upsert'
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.connect(...)
@@ -621,16 +631,6 @@ tables:
update-mode: append # otherwise: "retract" or "upsert"
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyTable (
- ...
-) WITH (
- 'update-mode' = 'append' -- otherwise: 'retract' or 'upsert'
-)
-{% endhighlight %}
-</div>
</div>
See also the [general streaming concepts documentation](streaming/dynamic_tables.html#continuous-queries) for more information.
@@ -655,6 +655,20 @@ Please note that not all connectors are available in both batch and streaming ye
The file system connector allows for reading and writing from a local or distributed filesystem. A filesystem can be defined as:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'connector.type' = 'filesystem', -- required: specify to connector type
+ 'connector.path' = 'file:///path/to/whatever', -- required: path to a file or directory
+ 'format.type' = '...', -- required: file system connector requires to specify a format,
+ ... -- currently only 'csv' format is supported.
+ -- Please refer to Table Formats section for more details.
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.connect(
@@ -689,20 +703,6 @@ format: # required: file system connector requires
# Please refer to Table Formats section for more details.
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'connector.type' = 'filesystem', -- required: specify to connector type
- 'connector.path' = 'file:///path/to/whatever', -- required: path to a file or directory
- 'format.type' = '...', -- required: file system connector requires to specify a format,
- ... -- currently only 'csv' format is supported.
- -- Please refer to Table Formats section for more details.
-)
-{% endhighlight %}
-</div>
</div>
The file system connector itself is included in Flink and does not require an additional dependency. A corresponding format needs to be specified for reading and writing rows from and to a file system.
@@ -720,6 +720,44 @@ The file system connector itself is included in Flink and does not require an ad
The Kafka connector allows for reading and writing from and to an Apache Kafka topic. It can be defined as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'connector.type' = 'kafka',
+
+ 'connector.version' = '0.11', -- required: valid connector versions are
+ -- "0.8", "0.9", "0.10", "0.11", and "universal"
+
+ 'connector.topic' = 'topic_name', -- required: topic name from which the table is read
+
+ 'connector.properties.zookeeper.connect' = 'localhost:2181', -- required: specify the ZooKeeper connection string
+ 'connector.properties.bootstrap.servers' = 'localhost:9092', -- required: specify the Kafka server connection string
+ 'connector.properties.group.id' = 'testGroup', --optional: required in Kafka consumer, specify consumer group
+ 'connector.startup-mode' = 'earliest-offset', -- optional: valid modes are "earliest-offset",
+ -- "latest-offset", "group-offsets",
+ -- or "specific-offsets"
+
+ -- optional: used in case of startup mode with specific offsets
+ 'connector.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300',
+
+ 'connector.sink-partitioner' = '...', -- optional: output partitioning from Flink's partitions
+ -- into Kafka's partitions valid are "fixed"
+ -- (each Flink partition ends up in at most one Kafka partition),
+ -- "round-robin" (a Flink partition is distributed to
+ -- Kafka partitions round-robin)
+ -- "custom" (use a custom FlinkKafkaPartitioner subclass)
+ -- optional: used in case of sink partitioner custom
+ 'connector.sink-partitioner-class' = 'org.mycompany.MyPartitioner',
+
+ 'format.type' = '...', -- required: Kafka connector requires to specify a format,
+ ... -- the supported formats are 'csv', 'json' and 'avro'.
+ -- Please refer to Table Formats section for more details.
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.connect(
@@ -806,44 +844,6 @@ connector:
# Please refer to Table Formats section for more details.
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'connector.type' = 'kafka',
-
- 'connector.version' = '0.11', -- required: valid connector versions are
- -- "0.8", "0.9", "0.10", "0.11", and "universal"
-
- 'connector.topic' = 'topic_name', -- required: topic name from which the table is read
-
- 'connector.properties.zookeeper.connect' = 'localhost:2181', -- required: specify the ZooKeeper connection string
- 'connector.properties.bootstrap.servers' = 'localhost:9092', -- required: specify the Kafka server connection string
- 'connector.properties.group.id' = 'testGroup', --optional: required in Kafka consumer, specify consumer group
- 'connector.startup-mode' = 'earliest-offset', -- optional: valid modes are "earliest-offset",
- -- "latest-offset", "group-offsets",
- -- or "specific-offsets"
-
- -- optional: used in case of startup mode with specific offsets
- 'connector.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300',
-
- 'connector.sink-partitioner' = '...', -- optional: output partitioning from Flink's partitions
- -- into Kafka's partitions valid are "fixed"
- -- (each Flink partition ends up in at most one Kafka partition),
- -- "round-robin" (a Flink partition is distributed to
- -- Kafka partitions round-robin)
- -- "custom" (use a custom FlinkKafkaPartitioner subclass)
- -- optional: used in case of sink partitioner custom
- 'connector.sink-partitioner-class' = 'org.mycompany.MyPartitioner',
-
- 'format.type' = '...', -- required: Kafka connector requires to specify a format,
- ... -- the supported formats are 'csv', 'json' and 'avro'.
- -- Please refer to Table Formats section for more details.
-)
-{% endhighlight %}
-</div>
</div>
**Specify the start reading position:** By default, the Kafka source will start reading data from the committed group offsets in Zookeeper or Kafka brokers. You can specify other start positions, which correspond to the configurations in section [Kafka Consumers Start Position Configuration]({{ site.baseurl }}/dev/connectors/kafka.html#kafka-consumers-start-position-configuration).
@@ -875,29 +875,91 @@ For append-only queries, the connector can also operate in [append mode](#update
The connector can be defined as follows:
<div class="codetabs" markdown="1">
-<div data-lang="Java/Scala" markdown="1">
-{% highlight java %}
-.connect(
- new Elasticsearch()
- .version("6") // required: valid connector versions are "6"
- .host("localhost", 9200, "http") // required: one or more Elasticsearch hosts to connect to
- .index("MyUsers") // required: Elasticsearch index
- .documentType("user") // required: Elasticsearch document type
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'connector.type' = 'elasticsearch', -- required: specify this table type is elasticsearch
+
+ 'connector.version' = '6', -- required: valid connector versions are "6"
+
+ 'connector.hosts' = 'http://host_name:9092;http://host_name:9093', -- required: one or more Elasticsearch hosts to connect to
- .keyDelimiter("$") // optional: delimiter for composite keys ("_" by default)
- // e.g., "$" would result in IDs "KEY1$KEY2$KEY3"
- .keyNullLiteral("n/a") // optional: representation for null fields in keys ("null" by default)
+ 'connector.index' = 'MyUsers', -- required: Elasticsearch index
- // optional: failure handling strategy in case a request to Elasticsearch fails (fail by default)
- .failureHandlerFail() // optional: throws an exception if a request fails and causes a job failure
- .failureHandlerIgnore() // or ignores failures and drops the request
- .failureHandlerRetryRejected() // or re-adds requests that have failed due to queue capacity saturation
- .failureHandlerCustom(...) // or custom failure handling with a ActionRequestFailureHandler subclass
+ 'connector.document-type' = 'user', -- required: Elasticsearch document type
- // optional: configure how to buffer elements before sending them in bulk to the cluster for efficiency
- .disableFlushOnCheckpoint() // optional: disables flushing on checkpoint (see notes below!)
- .bulkFlushMaxActions(42) // optional: maximum number of actions to buffer for each bulk request
- .bulkFlushMaxSize("42 mb") // optional: maximum size of buffered actions in bytes per bulk request
+ 'update-mode' = 'append', -- optional: update mode when used as table sink.
+
+ 'connector.key-delimiter' = '$', -- optional: delimiter for composite keys ("_" by default)
+ -- e.g., "$" would result in IDs "KEY1$KEY2$KEY3"
+
+ 'connector.key-null-literal' = 'n/a', -- optional: representation for null fields in keys ("null" by default)
+
+ 'connector.failure-handler' = '...', -- optional: failure handling strategy in case a request to
+ -- Elasticsearch fails ("fail" by default).
+ -- valid strategies are
+ -- "fail" (throws an exception if a request fails and
+ -- thus causes a job failure),
+ -- "ignore" (ignores failures and drops the request),
+ -- "retry-rejected" (re-adds requests that have failed due
+ -- to queue capacity saturation),
+ -- or "custom" for failure handling with a
+ -- ActionRequestFailureHandler subclass
+
+ -- optional: configure how to buffer elements before sending them in bulk to the cluster for efficiency
+ 'connector.flush-on-checkpoint' = 'true', -- optional: disables flushing on checkpoint (see notes below!)
+ -- ("true" by default)
+ 'connector.bulk-flush.max-actions' = '42', -- optional: maximum number of actions to buffer
+ -- for each bulk request
+ 'connector.bulk-flush.max-size' = '42 mb', -- optional: maximum size of buffered actions in bytes
+ -- per bulk request
+ -- (only MB granularity is supported)
+ 'connector.bulk-flush.interval' = '60000', -- optional: bulk flush interval (in milliseconds)
+ 'connector.bulk-flush.back-off.type' = '...', -- optional: backoff strategy ("disabled" by default)
+ -- valid strategies are "disabled", "constant",
+ -- or "exponential"
+ 'connector.bulk-flush.back-off.max-retries' = '3', -- optional: maximum number of retries
+ 'connector.bulk-flush.back-off.delay' = '30000', -- optional: delay between each backoff attempt
+ -- (in milliseconds)
+
+ -- optional: connection properties to be used during REST communication to Elasticsearch
+ 'connector.connection-max-retry-timeout' = '3', -- optional: maximum timeout (in milliseconds)
+ -- between retries
+ 'connector.connection-path-prefix' = '/v1' -- optional: prefix string to be added to every
+ -- REST communication
+
+ 'format.type' = '...', -- required: Elasticsearch connector requires to specify a format,
+ ... -- currently only 'json' format is supported.
+ -- Please refer to Table Formats section for more details.
+)
+{% endhighlight %}
+</div>
+
+<div data-lang="Java/Scala" markdown="1">
+{% highlight java %}
+.connect(
+ new Elasticsearch()
+ .version("6") // required: valid connector versions are "6"
+ .host("localhost", 9200, "http") // required: one or more Elasticsearch hosts to connect to
+ .index("MyUsers") // required: Elasticsearch index
+ .documentType("user") // required: Elasticsearch document type
+
+ .keyDelimiter("$") // optional: delimiter for composite keys ("_" by default)
+ // e.g., "$" would result in IDs "KEY1$KEY2$KEY3"
+ .keyNullLiteral("n/a") // optional: representation for null fields in keys ("null" by default)
+
+ // optional: failure handling strategy in case a request to Elasticsearch fails (fail by default)
+ .failureHandlerFail() // optional: throws an exception if a request fails and causes a job failure
+ .failureHandlerIgnore() // or ignores failures and drops the request
+ .failureHandlerRetryRejected() // or re-adds requests that have failed due to queue capacity saturation
+ .failureHandlerCustom(...) // or custom failure handling with a ActionRequestFailureHandler subclass
+
+ // optional: configure how to buffer elements before sending them in bulk to the cluster for efficiency
+ .disableFlushOnCheckpoint() // optional: disables flushing on checkpoint (see notes below!)
+ .bulkFlushMaxActions(42) // optional: maximum number of actions to buffer for each bulk request
+ .bulkFlushMaxSize("42 mb") // optional: maximum size of buffered actions in bytes per bulk request
// (only MB granularity is supported)
.bulkFlushInterval(60000L) // optional: bulk flush interval (in milliseconds)
@@ -1000,68 +1062,6 @@ connector:
# Please refer to Table Formats section for more details.
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'connector.type' = 'elasticsearch', -- required: specify this table type is elasticsearch
-
- 'connector.version' = '6', -- required: valid connector versions are "6"
-
- 'connector.hosts' = 'http://host_name:9092;http://host_name:9093', -- required: one or more Elasticsearch hosts to connect to
-
- 'connector.index' = 'MyUsers', -- required: Elasticsearch index
-
- 'connector.document-type' = 'user', -- required: Elasticsearch document type
-
- 'update-mode' = 'append', -- optional: update mode when used as table sink.
-
- 'connector.key-delimiter' = '$', -- optional: delimiter for composite keys ("_" by default)
- -- e.g., "$" would result in IDs "KEY1$KEY2$KEY3"
-
- 'connector.key-null-literal' = 'n/a', -- optional: representation for null fields in keys ("null" by default)
-
- 'connector.failure-handler' = '...', -- optional: failure handling strategy in case a request to
- -- Elasticsearch fails ("fail" by default).
- -- valid strategies are
- -- "fail" (throws an exception if a request fails and
- -- thus causes a job failure),
- -- "ignore" (ignores failures and drops the request),
- -- "retry-rejected" (re-adds requests that have failed due
- -- to queue capacity saturation),
- -- or "custom" for failure handling with a
- -- ActionRequestFailureHandler subclass
-
- -- optional: configure how to buffer elements before sending them in bulk to the cluster for efficiency
- 'connector.flush-on-checkpoint' = 'true', -- optional: disables flushing on checkpoint (see notes below!)
- -- ("true" by default)
- 'connector.bulk-flush.max-actions' = '42', -- optional: maximum number of actions to buffer
- -- for each bulk request
- 'connector.bulk-flush.max-size' = '42 mb', -- optional: maximum size of buffered actions in bytes
- -- per bulk request
- -- (only MB granularity is supported)
- 'connector.bulk-flush.interval' = '60000', -- optional: bulk flush interval (in milliseconds)
- 'connector.bulk-flush.back-off.type' = '...', -- optional: backoff strategy ("disabled" by default)
- -- valid strategies are "disabled", "constant",
- -- or "exponential"
- 'connector.bulk-flush.back-off.max-retries' = '3', -- optional: maximum number of retries
- 'connector.bulk-flush.back-off.delay' = '30000', -- optional: delay between each backoff attempt
- -- (in milliseconds)
-
- -- optional: connection properties to be used during REST communication to Elasticsearch
- 'connector.connection-max-retry-timeout' = '3', -- optional: maximum timeout (in milliseconds)
- -- between retries
- 'connector.connection-path-prefix' = '/v1' -- optional: prefix string to be added to every
- -- REST communication
-
- 'format.type' = '...', -- required: Elasticsearch connector requires to specify a format,
- ... -- currently only 'json' format is supported.
- -- Please refer to Table Formats section for more details.
-)
-{% endhighlight %}
-</div>
</div>
**Bulk flushing:** For more information about characteristics of the optional flushing parameters see the [corresponding low-level documentation]({{ site.baseurl }}/dev/connectors/elasticsearch.html).
@@ -1091,6 +1091,38 @@ For append-only queries, the connector can also operate in [append mode](#update
The connector can be defined as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ hbase_rowkey_name rowkey_type,
+ hbase_column_family_name1 ROW<...>,
+ hbase_column_family_name2 ROW<...>
+) WITH (
+ 'connector.type' = 'hbase', -- required: specify this table type is hbase
+
+ 'connector.version' = '1.4.3', -- required: valid connector versions are "1.4.3"
+
+ 'connector.table-name' = 'hbase_table_name', -- required: hbase table name
+
+ 'connector.zookeeper.quorum' = 'localhost:2181', -- required: HBase Zookeeper quorum configuration
+ 'connector.zookeeper.znode.parent' = '/test', -- optional: the root dir in Zookeeper for HBase cluster.
+ -- The default value is "/hbase".
+
+ 'connector.write.buffer-flush.max-size' = '10mb', -- optional: writing option, determines how many size in memory of buffered
+ -- rows to insert per round trip. This can help performance on writing to JDBC
+ -- database. The default value is "2mb".
+
+ 'connector.write.buffer-flush.max-rows' = '1000', -- optional: writing option, determines how many rows to insert per round trip.
+ -- This can help performance on writing to JDBC database. No default value,
+ -- i.e. the default flushing is not depends on the number of buffered rows.
+
+ 'connector.write.buffer-flush.interval' = '2s', -- optional: writing option, sets a flush interval flushing buffered requesting
+ -- if the interval passes, in milliseconds. Default value is "0s", which means
+ -- no asynchronous flush thread will be scheduled.
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.connect(
@@ -1137,38 +1169,6 @@ connector:
# no asynchronous flush thread will be scheduled.
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- hbase_rowkey_name rowkey_type,
- hbase_column_family_name1 ROW<...>,
- hbase_column_family_name2 ROW<...>
-) WITH (
- 'connector.type' = 'hbase', -- required: specify this table type is hbase
-
- 'connector.version' = '1.4.3', -- required: valid connector versions are "1.4.3"
-
- 'connector.table-name' = 'hbase_table_name', -- required: hbase table name
-
- 'connector.zookeeper.quorum' = 'localhost:2181', -- required: HBase Zookeeper quorum configuration
- 'connector.zookeeper.znode.parent' = '/test', -- optional: the root dir in Zookeeper for HBase cluster.
- -- The default value is "/hbase".
-
- 'connector.write.buffer-flush.max-size' = '10mb', -- optional: writing option, determines how many size in memory of buffered
- -- rows to insert per round trip. This can help performance on writing to JDBC
- -- database. The default value is "2mb".
-
- 'connector.write.buffer-flush.max-rows' = '1000', -- optional: writing option, determines how many rows to insert per round trip.
- -- This can help performance on writing to JDBC database. No default value,
- -- i.e. the default flushing is not depends on the number of buffered rows.
-
- 'connector.write.buffer-flush.interval' = '2s', -- optional: writing option, sets a flush interval flushing buffered requesting
- -- if the interval passes, in milliseconds. Default value is "0s", which means
- -- no asynchronous flush thread will be scheduled.
-)
-{% endhighlight %}
-</div>
</div>
**Columns:** All the column families in HBase table must be declared as `ROW` type, the field name maps to the column family name, and the nested field names map to the column qualifier names. There is no need to declare all the families and qualifiers in the schema, users can declare what's necessary. Except the `ROW` type fields, the only one field of atomic type (e.g. `STRING`, `BIGINT`) will be recognized as row key of the table. There's no constraints on the name of row key field.
@@ -1206,52 +1206,6 @@ To use JDBC connector, need to choose an actual driver to use. Here are drivers
The connector can be defined as follows:
<div class="codetabs" markdown="1">
-<div data-lang="YAML" markdown="1">
-{% highlight yaml %}
-connector:
- type: jdbc
- url: "jdbc:mysql://localhost:3306/flink-test" # required: JDBC DB url
- table: "jdbc_table_name" # required: jdbc table name
- driver: "com.mysql.jdbc.Driver" # optional: the class name of the JDBC driver to use to connect to this URL.
- # If not set, it will automatically be derived from the URL.
-
- username: "name" # optional: jdbc user name and password
- password: "password"
-
- read: # scan options, optional, used when reading from table
- partition: # These options must all be specified if any of them is specified. In addition, partition.num must be specified. They
- # describe how to partition the table when reading in parallel from multiple tasks. partition.column must be a numeric,
- # date, or timestamp column from the table in question. Notice that lowerBound and upperBound are just used to decide
- # the partition stride, not for filtering the rows in table. So all rows in the table will be partitioned and returned.
- # This option applies only to reading.
- column: "column_name" # optional, name of the column used for partitioning the input.
- num: 50 # optional, the number of partitions.
- lower-bound: 500 # optional, the smallest value of the first partition.
- upper-bound: 1000 # optional, the largest value of the last partition.
- fetch-size: 100 # optional, Gives the reader a hint as to the number of rows that should be fetched
- # from the database when reading per round trip. If the value specified is zero, then
- # the hint is ignored. The default value is zero.
-
- lookup: # lookup options, optional, used in temporary join
- cache:
- max-rows: 5000 # optional, max number of rows of lookup cache, over this value, the oldest rows will
- # be eliminated. "cache.max-rows" and "cache.ttl" options must all be specified if any
- # of them is specified. Cache is not enabled as default.
- ttl: "10s" # optional, the max time to live for each rows in lookup cache, over this time, the oldest rows
- # will be expired. "cache.max-rows" and "cache.ttl" options must all be specified if any of
- # them is specified. Cache is not enabled as default.
- max-retries: 3 # optional, max retry times if lookup database failed
-
- write: # sink options, optional, used when writing into table
- flush:
- max-rows: 5000 # optional, flush max size (includes all append, upsert and delete records),
- # over this number of records, will flush data. The default value is "5000".
- interval: "2s" # optional, flush interval mills, over this time, asynchronous threads will flush data.
- # The default value is "0s", which means no asynchronous flush thread will be scheduled.
- max-retries: 3 # optional, max retry times if writing records to database failed.
-{% endhighlight %}
-</div>
-
<div data-lang="DDL" markdown="1">
{% highlight sql %}
CREATE TABLE MyUserTable (
@@ -1303,6 +1257,52 @@ CREATE TABLE MyUserTable (
)
{% endhighlight %}
</div>
+
+<div data-lang="YAML" markdown="1">
+{% highlight yaml %}
+connector:
+ type: jdbc
+ url: "jdbc:mysql://localhost:3306/flink-test" # required: JDBC DB url
+ table: "jdbc_table_name" # required: jdbc table name
+ driver: "com.mysql.jdbc.Driver" # optional: the class name of the JDBC driver to use to connect to this URL.
+ # If not set, it will automatically be derived from the URL.
+
+ username: "name" # optional: jdbc user name and password
+ password: "password"
+
+ read: # scan options, optional, used when reading from table
+ partition: # These options must all be specified if any of them is specified. In addition, partition.num must be specified. They
+ # describe how to partition the table when reading in parallel from multiple tasks. partition.column must be a numeric,
+ # date, or timestamp column from the table in question. Notice that lowerBound and upperBound are just used to decide
+ # the partition stride, not for filtering the rows in table. So all rows in the table will be partitioned and returned.
+ # This option applies only to reading.
+ column: "column_name" # optional, name of the column used for partitioning the input.
+ num: 50 # optional, the number of partitions.
+ lower-bound: 500 # optional, the smallest value of the first partition.
+ upper-bound: 1000 # optional, the largest value of the last partition.
+ fetch-size: 100 # optional, Gives the reader a hint as to the number of rows that should be fetched
+ # from the database when reading per round trip. If the value specified is zero, then
+ # the hint is ignored. The default value is zero.
+
+ lookup: # lookup options, optional, used in temporary join
+ cache:
+ max-rows: 5000 # optional, max number of rows of lookup cache, over this value, the oldest rows will
+ # be eliminated. "cache.max-rows" and "cache.ttl" options must all be specified if any
+ # of them is specified. Cache is not enabled as default.
+ ttl: "10s" # optional, the max time to live for each rows in lookup cache, over this time, the oldest rows
+ # will be expired. "cache.max-rows" and "cache.ttl" options must all be specified if any of
+ # them is specified. Cache is not enabled as default.
+ max-retries: 3 # optional, max retry times if lookup database failed
+
+ write: # sink options, optional, used when writing into table
+ flush:
+ max-rows: 5000 # optional, flush max size (includes all append, upsert and delete records),
+ # over this number of records, will flush data. The default value is "5000".
+ interval: "2s" # optional, flush interval mills, over this time, asynchronous threads will flush data.
+ # The default value is "0s", which means no asynchronous flush thread will be scheduled.
+ max-retries: 3 # optional, max retry times if writing records to database failed.
+{% endhighlight %}
+</div>
</div>
**Upsert sink:** Flink automatically extracts valid keys from a query. For example, a query `SELECT a, b, c FROM t GROUP BY a, b` defines a composite key of the fields `a` and `b`. If a JDBC table is used as upsert sink, please make sure keys of the query is one of the unique key sets or primary key of the underlying database. This can guarantee the output result is as expected.
@@ -1351,6 +1351,36 @@ schema is interpreted as a field renaming in the format.
The CSV format can be used as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'format.type' = 'csv', -- required: specify the schema type
+
+ 'format.fields.0.name' = 'lon', -- optional: define the schema explicitly using type information.
+ 'format.fields.0.data-type' = 'FLOAT', -- This overrides default behavior that uses table's schema as format schema.
+ 'format.fields.1.name' = 'rideTime',
+ 'format.fields.1.data-type' = 'TIMESTAMP(3)',
+
+ 'format.field-delimiter' = ';', -- optional: field delimiter character (',' by default)
+ 'format.line-delimiter' = '\r\n', -- optional: line delimiter ("\n" by default; otherwise
+ -- "\r" or "\r\n" are allowed)
+ 'format.quote-character' = '''', -- optional: quote character for enclosing field values ('"' by default)
+ 'format.allow-comments' = 'true', -- optional: ignores comment lines that start with "#"
+ -- (disabled by default);
+ -- if enabled, make sure to also ignore parse errors to allow empty rows
+ 'format.ignore-parse-errors' = 'true', -- optional: skip fields and rows with parse errors instead of failing;
+ -- fields are set to null in case of errors
+ 'format.array-element-delimiter' = '|', -- optional: the array element delimiter string for separating
+ -- array and row element values (";" by default)
+ 'format.escape-character' = '\\', -- optional: escape character for escaping values (disabled by default)
+ 'format.null-literal' = 'n/a' -- optional: null literal string that is interpreted as a
+ -- null value (disabled by default)
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.withFormat(
@@ -1427,36 +1457,6 @@ format:
# null value (disabled by default)
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'format.type' = 'csv', -- required: specify the schema type
-
- 'format.fields.0.name' = 'lon', -- optional: define the schema explicitly using type information.
- 'format.fields.0.data-type' = 'FLOAT', -- This overrides default behavior that uses table's schema as format schema.
- 'format.fields.1.name' = 'rideTime',
- 'format.fields.1.data-type' = 'TIMESTAMP(3)',
-
- 'format.field-delimiter' = ';', -- optional: field delimiter character (',' by default)
- 'format.line-delimiter' = '\r\n', -- optional: line delimiter ("\n" by default; otherwise
- -- "\r" or "\r\n" are allowed)
- 'format.quote-character' = '''', -- optional: quote character for enclosing field values ('"' by default)
- 'format.allow-comments' = 'true', -- optional: ignores comment lines that start with "#"
- -- (disabled by default);
- -- if enabled, make sure to also ignore parse errors to allow empty rows
- 'format.ignore-parse-errors' = 'true', -- optional: skip fields and rows with parse errors instead of failing;
- -- fields are set to null in case of errors
- 'format.array-element-delimiter' = '|', -- optional: the array element delimiter string for separating
- -- array and row element values (";" by default)
- 'format.escape-character' = '\\', -- optional: escape character for escaping values (disabled by default)
- 'format.null-literal' = 'n/a' -- optional: null literal string that is interpreted as a
- -- null value (disabled by default)
-)
-{% endhighlight %}
-</div>
</div>
The following table lists supported types that can be read and written:
@@ -1519,6 +1519,36 @@ If the format schema is equal to the table schema, the schema can also be automa
The JSON format can be used as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'format.type' = 'json', -- required: specify the format type
+ 'format.fail-on-missing-field' = 'true' -- optional: flag whether to fail if a field is missing or not, false by default
+
+ 'format.fields.0.name' = 'lon', -- optional: define the schema explicitly using type information.
+ 'format.fields.0.data-type' = 'FLOAT', -- This overrides default behavior that uses table's schema as format schema.
+ 'format.fields.1.name' = 'rideTime',
+ 'format.fields.1.data-type' = 'TIMESTAMP(3)',
+
+ 'format.json-schema' = -- or by using a JSON schema which parses to DECIMAL and TIMESTAMP.
+ '{ -- This also overrides the default behavior.
+ "type": "object",
+ "properties": {
+ "lon": {
+ "type": "number"
+ },
+ "rideTime": {
+ "type": "string",
+ "format": "date-time"
+ }
+ }
+ }'
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.withFormat(
@@ -1603,36 +1633,6 @@ format:
}
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'format.type' = 'json', -- required: specify the format type
- 'format.fail-on-missing-field' = 'true' -- optional: flag whether to fail if a field is missing or not, false by default
-
- 'format.fields.0.name' = 'lon', -- optional: define the schema explicitly using type information.
- 'format.fields.0.data-type' = 'FLOAT', -- This overrides default behavior that uses table's schema as format schema.
- 'format.fields.1.name' = 'rideTime',
- 'format.fields.1.data-type' = 'TIMESTAMP(3)',
-
- 'format.json-schema' = -- or by using a JSON schema which parses to DECIMAL and TIMESTAMP.
- '{ -- This also overrides the default behavior.
- "type": "object",
- "properties": {
- "lon": {
- "type": "number"
- },
- "rideTime": {
- "type": "string",
- "format": "date-time"
- }
- }
- }'
-)
-{% endhighlight %}
-</div>
</div>
The following table shows the mapping of JSON schema types to Flink SQL types:
@@ -1714,6 +1714,27 @@ The [Apache Avro](https://avro.apache.org/) format allows to read and write Avro
The Avro format can be used as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'format.type' = 'avro', -- required: specify the schema type
+ 'format.record-class' = 'org.organization.types.User', -- required: define the schema either by using an Avro specific record class
+
+ 'format.avro-schema' = -- or by using an Avro schema
+ '{
+ "type": "record",
+ "name": "test",
+ "fields" : [
+ {"name": "a", "type": "long"},
+ {"name": "b", "type": "string"}
+ ]
+ }'
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.withFormat(
@@ -1780,27 +1801,6 @@ format:
}
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'format.type' = 'avro', -- required: specify the schema type
- 'format.record-class' = 'org.organization.types.User', -- required: define the schema either by using an Avro specific record class
-
- 'format.avro-schema' = -- or by using an Avro schema
- '{
- "type": "record",
- "name": "test",
- "fields" : [
- {"name": "a", "type": "long"},
- {"name": "b", "type": "string"}
- ]
- }'
-)
-{% endhighlight %}
-</div>
</div>
Avro types are mapped to the corresponding SQL data types. Union types are only supported for specifying nullability otherwise they are converted to an `ANY` type. The following table shows the mapping:
@@ -1844,6 +1844,28 @@ replaced by a proper RFC-compliant version. Use the RFC-compliant CSV format whe
Use the old one for stream/batch filesystem operations for now.
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'format.type' = 'csv', -- required: specify the schema type
+
+ 'format.fields.0.name' = 'lon', -- optional: declare ordered format fields explicitly. This will overrides
+ 'format.fields.0.data-type' = 'STRING', -- the default behavior that uses table's schema as format schema.
+ 'format.fields.1.name' = 'rideTime',
+ 'format.fields.1.data-type' = 'TIMESTAMP(3)',
+
+ 'format.field-delimiter' = ',', -- optional: string delimiter "," by default
+ 'format.line-delimiter' = '\n', -- optional: string delimiter "\n" by default
+ 'format.quote-character' = '"', -- optional: single character for string values, empty by default
+ 'format.comment-prefix' = '#', -- optional: string to indicate comments, empty by default
+ 'format.ignore-first-line' = 'false', -- optional: boolean flag to ignore the first line, by default it is not skipped
+ 'format.ignore-parse-errors' = 'true' -- optional: skip records with parse error instead of failing by default
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.withFormat(
@@ -1893,28 +1915,6 @@ format:
ignore-parse-errors: true # optional: skip records with parse error instead of failing by default
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'format.type' = 'csv', -- required: specify the schema type
-
- 'format.fields.0.name' = 'lon', -- optional: declare ordered format fields explicitly. This will overrides
- 'format.fields.0.data-type' = 'STRING', -- the default behavior that uses table's schema as format schema.
- 'format.fields.1.name' = 'rideTime',
- 'format.fields.1.data-type' = 'TIMESTAMP(3)',
-
- 'format.field-delimiter' = ',', -- optional: string delimiter "," by default
- 'format.line-delimiter' = '\n', -- optional: string delimiter "\n" by default
- 'format.quote-character' = '"', -- optional: single character for string values, empty by default
- 'format.comment-prefix' = '#', -- optional: string to indicate comments, empty by default
- 'format.ignore-first-line' = 'false', -- optional: boolean flag to ignore the first line, by default it is not skipped
- 'format.ignore-parse-errors' = 'true' -- optional: skip records with parse error instead of failing by default
-)
-{% endhighlight %}
-</div>
</div>
The old CSV format is included in Flink and does not require additional dependencies.
diff --git a/docs/dev/table/connect.zh.md b/docs/dev/table/connect.zh.md
index 31f2732..f4c8e39 100644
--- a/docs/dev/table/connect.zh.md
+++ b/docs/dev/table/connect.zh.md
@@ -92,6 +92,21 @@ The **table schema** defines the schema of a table that is exposed to SQL querie
The subsequent sections will cover each definition part ([connector](connect.html#table-connectors), [format](connect.html#table-formats), and [schema](connect.html#table-schema)) in more detail. The following example shows how to pass them:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+tableEnvironment.sqlUpdate(
+ "CREATE TABLE MyTable (\n" +
+ " ... -- declare table schema \n" +
+ ") WITH (\n" +
+ " 'connector.type' = '...', -- declare connector specific properties\n" +
+ " ...\n" +
+ " 'update-mode' = 'append', -- declare update mode\n" +
+ " 'format.type' = '...', -- declare format specific properties\n" +
+ " ...\n" +
+ ")");
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
tableEnvironment
@@ -124,21 +139,6 @@ format: ...
schema: ...
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-tableEnvironment.sqlUpdate(
- "CREATE TABLE MyTable (\n" +
- " ... -- declare table schema \n" +
- ") WITH (\n" +
- " 'connector.type' = '...', -- declare connector specific properties\n" +
- " ...\n" +
- " 'update-mode' = 'append', -- declare update mode\n" +
- " 'format.type' = '...', -- declare format specific properties\n" +
- " ...\n" +
- ")");
-{% endhighlight %}
-</div>
</div>
The table's type (`source`, `sink`, or `both`) determines how a table is registered. In case of table type `both`, both a table source and table sink are registered under the same name. Logically, this means that we can both read and write to such a table similarly to a table in a regular DBMS.
@@ -148,6 +148,41 @@ For streaming queries, an [update mode](connect.html#update-mode) declares how t
The following code shows a full example of how to connect to Kafka for reading Avro records.
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ -- declare the schema of the table
+ `user` BIGINT,
+ message STRING,
+ ts STRING
+) WITH (
+ -- declare the external system to connect to
+ 'connector.type' = 'kafka',
+ 'connector.version' = '0.10',
+ 'connector.topic' = 'topic_name',
+ 'connector.startup-mode' = 'earliest-offset',
+ 'connector.properties.zookeeper.connect' = 'localhost:2181',
+ 'connector.properties.bootstrap.servers' = 'localhost:9092',
+
+ -- specify the update-mode for streaming tables
+ 'update-mode' = 'append',
+
+ -- declare a format for this system
+ 'format.type' = 'avro',
+ 'format.avro-schema' = '{
+ "namespace": "org.myorganization",
+ "type": "record",
+ "name": "UserMessage",
+ "fields": [
+ {"name": "ts", "type": "string"},
+ {"name": "user", "type": "long"},
+ {"name": "message", "type": ["string", "null"]}
+ ]
+ }'
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
tableEnvironment
@@ -285,41 +320,6 @@ tables:
data-type: STRING
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- -- declare the schema of the table
- `user` BIGINT,
- message STRING,
- ts STRING
-) WITH (
- -- declare the external system to connect to
- 'connector.type' = 'kafka',
- 'connector.version' = '0.10',
- 'connector.topic' = 'topic_name',
- 'connector.startup-mode' = 'earliest-offset',
- 'connector.properties.zookeeper.connect' = 'localhost:2181',
- 'connector.properties.bootstrap.servers' = 'localhost:9092',
-
- -- specify the update-mode for streaming tables
- 'update-mode' = 'append',
-
- -- declare a format for this system
- 'format.type' = 'avro',
- 'format.avro-schema' = '{
- "namespace": "org.myorganization",
- "type": "record",
- "name": "UserMessage",
- "fields": [
- {"name": "ts", "type": "string"},
- {"name": "user", "type": "long"},
- {"name": "message", "type": ["string", "null"]}
- ]
- }'
-)
-{% endhighlight %}
-</div>
</div>
In both ways the desired connection properties are converted into normalized, string-based key-value pairs. So-called [table factories](sourceSinks.html#define-a-tablefactory) create configured table sources, table sinks, and corresponding formats from the key-value pairs. All table factories that can be found via Java's [Service Provider Interfaces (SPI)](https://docs.oracle.com/javase/tutorial/sound/SPI-intro.html) are taken into account when searching for exactly-one matching table factory.
@@ -600,6 +600,16 @@ For streaming queries, it is required to declare how to perform the [conversion
<span class="label label-danger">Attention</span> The documentation of each connector states which update modes are supported.
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyTable (
+ ...
+) WITH (
+ 'update-mode' = 'append' -- otherwise: 'retract' or 'upsert'
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.connect(...)
@@ -621,16 +631,6 @@ tables:
update-mode: append # otherwise: "retract" or "upsert"
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyTable (
- ...
-) WITH (
- 'update-mode' = 'append' -- otherwise: 'retract' or 'upsert'
-)
-{% endhighlight %}
-</div>
</div>
See also the [general streaming concepts documentation](streaming/dynamic_tables.html#continuous-queries) for more information.
@@ -655,6 +655,20 @@ Please note that not all connectors are available in both batch and streaming ye
The file system connector allows for reading and writing from a local or distributed filesystem. A filesystem can be defined as:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'connector.type' = 'filesystem', -- required: specify to connector type
+ 'connector.path' = 'file:///path/to/whatever', -- required: path to a file or directory
+ 'format.type' = '...', -- required: file system connector requires to specify a format,
+ ... -- currently only 'csv' format is supported.
+ -- Please refer to Table Formats section for more details.
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.connect(
@@ -689,20 +703,6 @@ format: # required: file system connector requires
# Please refer to Table Formats section for more details.
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'connector.type' = 'filesystem', -- required: specify to connector type
- 'connector.path' = 'file:///path/to/whatever', -- required: path to a file or directory
- 'format.type' = '...', -- required: file system connector requires to specify a format,
- ... -- currently only 'csv' format is supported.
- -- Please refer to Table Formats section for more details.
-)
-{% endhighlight %}
-</div>
</div>
The file system connector itself is included in Flink and does not require an additional dependency. A corresponding format needs to be specified for reading and writing rows from and to a file system.
@@ -720,6 +720,44 @@ The file system connector itself is included in Flink and does not require an ad
The Kafka connector allows for reading and writing from and to an Apache Kafka topic. It can be defined as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'connector.type' = 'kafka',
+
+ 'connector.version' = '0.11', -- required: valid connector versions are
+ -- "0.8", "0.9", "0.10", "0.11", and "universal"
+
+ 'connector.topic' = 'topic_name', -- required: topic name from which the table is read
+
+ 'connector.properties.zookeeper.connect' = 'localhost:2181', -- required: specify the ZooKeeper connection string
+ 'connector.properties.bootstrap.servers' = 'localhost:9092', -- required: specify the Kafka server connection string
+ 'connector.properties.group.id' = 'testGroup', --optional: required in Kafka consumer, specify consumer group
+ 'connector.startup-mode' = 'earliest-offset', -- optional: valid modes are "earliest-offset",
+ -- "latest-offset", "group-offsets",
+ -- or "specific-offsets"
+
+ -- optional: used in case of startup mode with specific offsets
+ 'connector.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300',
+
+ 'connector.sink-partitioner' = '...', -- optional: output partitioning from Flink's partitions
+ -- into Kafka's partitions valid are "fixed"
+ -- (each Flink partition ends up in at most one Kafka partition),
+ -- "round-robin" (a Flink partition is distributed to
+ -- Kafka partitions round-robin)
+ -- "custom" (use a custom FlinkKafkaPartitioner subclass)
+ -- optional: used in case of sink partitioner custom
+ 'connector.sink-partitioner-class' = 'org.mycompany.MyPartitioner',
+
+ 'format.type' = '...', -- required: Kafka connector requires to specify a format,
+ ... -- the supported formats are 'csv', 'json' and 'avro'.
+ -- Please refer to Table Formats section for more details.
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.connect(
@@ -756,7 +794,7 @@ The Kafka connector allows for reading and writing from and to an Apache Kafka t
.version("0.11") # required: valid connector versions are
# "0.8", "0.9", "0.10", "0.11", and "universal"
.topic("...") # required: topic name from which the table is read
-
+
# optional: connector specific properties
.property("zookeeper.connect", "localhost:2181")
.property("bootstrap.servers", "localhost:9092")
@@ -794,56 +832,18 @@ connector:
startup-mode: ... # optional: valid modes are "earliest-offset", "latest-offset",
# "group-offsets", or "specific-offsets"
specific-offsets: partition:0,offset:42;partition:1,offset:300 # optional: used in case of startup mode with specific offsets
-
+
sink-partitioner: ... # optional: output partitioning from Flink's partitions into Kafka's partitions
# valid are "fixed" (each Flink partition ends up in at most one Kafka partition),
# "round-robin" (a Flink partition is distributed to Kafka partitions round-robin)
# "custom" (use a custom FlinkKafkaPartitioner subclass)
sink-partitioner-class: org.mycompany.MyPartitioner # optional: used in case of sink partitioner custom
-
+
format: # required: Kafka connector requires to specify a format,
... # the supported formats are "csv", "json" and "avro".
# Please refer to Table Formats section for more details.
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'connector.type' = 'kafka',
-
- 'connector.version' = '0.11', -- required: valid connector versions are
- -- "0.8", "0.9", "0.10", "0.11", and "universal"
-
- 'connector.topic' = 'topic_name', -- required: topic name from which the table is read
-
- 'connector.properties.zookeeper.connect' = 'localhost:2181', -- required: specify the ZooKeeper connection string
- 'connector.properties.bootstrap.servers' = 'localhost:9092', -- required: specify the Kafka server connection string
- 'connector.properties.group.id' = 'testGroup', --optional: required in Kafka consumer, specify consumer group
- 'connector.startup-mode' = 'earliest-offset', -- optional: valid modes are "earliest-offset",
- -- "latest-offset", "group-offsets",
- -- or "specific-offsets"
-
- -- optional: used in case of startup mode with specific offsets
- 'connector.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300',
-
- 'connector.sink-partitioner' = '...', -- optional: output partitioning from Flink's partitions
- -- into Kafka's partitions valid are "fixed"
- -- (each Flink partition ends up in at most one Kafka partition),
- -- "round-robin" (a Flink partition is distributed to
- -- Kafka partitions round-robin)
- -- "custom" (use a custom FlinkKafkaPartitioner subclass)
- -- optional: used in case of sink partitioner custom
- 'connector.sink-partitioner-class' = 'org.mycompany.MyPartitioner',
-
- 'format.type' = '...', -- required: Kafka connector requires to specify a format,
- ... -- the supported formats are 'csv', 'json' and 'avro'.
- -- Please refer to Table Formats section for more details.
-)
-{% endhighlight %}
-</div>
</div>
**Specify the start reading position:** By default, the Kafka source will start reading data from the committed group offsets in Zookeeper or Kafka brokers. You can specify other start positions, which correspond to the configurations in section [Kafka Consumers Start Position Configuration]({{ site.baseurl }}/dev/connectors/kafka.html#kafka-consumers-start-position-configuration).
@@ -875,45 +875,107 @@ For append-only queries, the connector can also operate in [append mode](#update
The connector can be defined as follows:
<div class="codetabs" markdown="1">
-<div data-lang="Java/Scala" markdown="1">
-{% highlight java %}
-.connect(
- new Elasticsearch()
- .version("6") // required: valid connector versions are "6"
- .host("localhost", 9200, "http") // required: one or more Elasticsearch hosts to connect to
- .index("MyUsers") // required: Elasticsearch index
- .documentType("user") // required: Elasticsearch document type
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'connector.type' = 'elasticsearch', -- required: specify this table type is elasticsearch
+
+ 'connector.version' = '6', -- required: valid connector versions are "6"
+
+ 'connector.hosts' = 'http://host_name:9092;http://host_name:9093', -- required: one or more Elasticsearch hosts to connect to
- .keyDelimiter("$") // optional: delimiter for composite keys ("_" by default)
- // e.g., "$" would result in IDs "KEY1$KEY2$KEY3"
- .keyNullLiteral("n/a") // optional: representation for null fields in keys ("null" by default)
+ 'connector.index' = 'MyUsers', -- required: Elasticsearch index
- // optional: failure handling strategy in case a request to Elasticsearch fails (fail by default)
- .failureHandlerFail() // optional: throws an exception if a request fails and causes a job failure
- .failureHandlerIgnore() // or ignores failures and drops the request
- .failureHandlerRetryRejected() // or re-adds requests that have failed due to queue capacity saturation
- .failureHandlerCustom(...) // or custom failure handling with a ActionRequestFailureHandler subclass
+ 'connector.document-type' = 'user', -- required: Elasticsearch document type
- // optional: configure how to buffer elements before sending them in bulk to the cluster for efficiency
- .disableFlushOnCheckpoint() // optional: disables flushing on checkpoint (see notes below!)
- .bulkFlushMaxActions(42) // optional: maximum number of actions to buffer for each bulk request
- .bulkFlushMaxSize("42 mb") // optional: maximum size of buffered actions in bytes per bulk request
- // (only MB granularity is supported)
- .bulkFlushInterval(60000L) // optional: bulk flush interval (in milliseconds)
+ 'update-mode' = 'append', -- optional: update mode when used as table sink.
- .bulkFlushBackoffConstant() // optional: use a constant backoff type
- .bulkFlushBackoffExponential() // or use an exponential backoff type
- .bulkFlushBackoffMaxRetries(3) // optional: maximum number of retries
- .bulkFlushBackoffDelay(30000L) // optional: delay between each backoff attempt (in milliseconds)
+ 'connector.key-delimiter' = '$', -- optional: delimiter for composite keys ("_" by default)
+ -- e.g., "$" would result in IDs "KEY1$KEY2$KEY3"
- // optional: connection properties to be used during REST communication to Elasticsearch
- .connectionMaxRetryTimeout(3) // optional: maximum timeout (in milliseconds) between retries
- .connectionPathPrefix("/v1") // optional: prefix string to be added to every REST communication
-)
+ 'connector.key-null-literal' = 'n/a', -- optional: representation for null fields in keys ("null" by default)
+
+ 'connector.failure-handler' = '...', -- optional: failure handling strategy in case a request to
+ -- Elasticsearch fails ("fail" by default).
+ -- valid strategies are
+ -- "fail" (throws an exception if a request fails and
+ -- thus causes a job failure),
+ -- "ignore" (ignores failures and drops the request),
+ -- "retry-rejected" (re-adds requests that have failed due
+ -- to queue capacity saturation),
+ -- or "custom" for failure handling with a
+ -- ActionRequestFailureHandler subclass
+
+ -- optional: configure how to buffer elements before sending them in bulk to the cluster for efficiency
+ 'connector.flush-on-checkpoint' = 'true', -- optional: disables flushing on checkpoint (see notes below!)
+ -- ("true" by default)
+ 'connector.bulk-flush.max-actions' = '42', -- optional: maximum number of actions to buffer
+ -- for each bulk request
+ 'connector.bulk-flush.max-size' = '42 mb', -- optional: maximum size of buffered actions in bytes
+ -- per bulk request
+ -- (only MB granularity is supported)
+ 'connector.bulk-flush.interval' = '60000', -- optional: bulk flush interval (in milliseconds)
+ 'connector.bulk-flush.back-off.type' = '...', -- optional: backoff strategy ("disabled" by default)
+ -- valid strategies are "disabled", "constant",
+ -- or "exponential"
+ 'connector.bulk-flush.back-off.max-retries' = '3', -- optional: maximum number of retries
+ 'connector.bulk-flush.back-off.delay' = '30000', -- optional: delay between each backoff attempt
+ -- (in milliseconds)
+
+ -- optional: connection properties to be used during REST communication to Elasticsearch
+ 'connector.connection-max-retry-timeout' = '3', -- optional: maximum timeout (in milliseconds)
+ -- between retries
+ 'connector.connection-path-prefix' = '/v1' -- optional: prefix string to be added to every
+ -- REST communication
+
+ 'format.type' = '...', -- required: Elasticsearch connector requires to specify a format,
+ ... -- currently only 'json' format is supported.
+ -- Please refer to Table Formats section for more details.
+)
+{% endhighlight %}
+</div>
+
+<div data-lang="Java/Scala" markdown="1">
+{% highlight java %}
+.connect(
+ new Elasticsearch()
+ .version("6") // required: valid connector versions are "6"
+ .host("localhost", 9200, "http") // required: one or more Elasticsearch hosts to connect to
+ .index("MyUsers") // required: Elasticsearch index
+ .documentType("user") // required: Elasticsearch document type
+
+ .keyDelimiter("$") // optional: delimiter for composite keys ("_" by default)
+ // e.g., "$" would result in IDs "KEY1$KEY2$KEY3"
+ .keyNullLiteral("n/a") // optional: representation for null fields in keys ("null" by default)
+
+ // optional: failure handling strategy in case a request to Elasticsearch fails (fail by default)
+ .failureHandlerFail() // optional: throws an exception if a request fails and causes a job failure
+ .failureHandlerIgnore() // or ignores failures and drops the request
+ .failureHandlerRetryRejected() // or re-adds requests that have failed due to queue capacity saturation
+ .failureHandlerCustom(...) // or custom failure handling with a ActionRequestFailureHandler subclass
+
+ // optional: configure how to buffer elements before sending them in bulk to the cluster for efficiency
+ .disableFlushOnCheckpoint() // optional: disables flushing on checkpoint (see notes below!)
+ .bulkFlushMaxActions(42) // optional: maximum number of actions to buffer for each bulk request
+ .bulkFlushMaxSize("42 mb") // optional: maximum size of buffered actions in bytes per bulk request
+ // (only MB granularity is supported)
+ .bulkFlushInterval(60000L) // optional: bulk flush interval (in milliseconds)
+
+ .bulkFlushBackoffConstant() // optional: use a constant backoff type
+ .bulkFlushBackoffExponential() // or use an exponential backoff type
+ .bulkFlushBackoffMaxRetries(3) // optional: maximum number of retries
+ .bulkFlushBackoffDelay(30000L) // optional: delay between each backoff attempt (in milliseconds)
+
+ // optional: connection properties to be used during REST communication to Elasticsearch
+ .connectionMaxRetryTimeout(3) // optional: maximum timeout (in milliseconds) between retries
+ .connectionPathPrefix("/v1") // optional: prefix string to be added to every REST communication
+)
.withFormat( // required: Elasticsearch connector requires to specify a format,
... // currently only Json format is supported.
// Please refer to Table Formats section for more details.
-)
+)
{% endhighlight %}
</div>
@@ -994,74 +1056,12 @@ connector:
# optional: connection properties to be used during REST communication to Elasticsearch
connection-max-retry-timeout: 3 # optional: maximum timeout (in milliseconds) between retries
connection-path-prefix: "/v1" # optional: prefix string to be added to every REST communication
-
+
format: # required: Elasticsearch connector requires to specify a format,
... # currently only "json" format is supported.
# Please refer to Table Formats section for more details.
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'connector.type' = 'elasticsearch', -- required: specify this table type is elasticsearch
-
- 'connector.version' = '6', -- required: valid connector versions are "6"
-
- 'connector.hosts' = 'http://host_name:9092;http://host_name:9093', -- required: one or more Elasticsearch hosts to connect to
-
- 'connector.index' = 'MyUsers', -- required: Elasticsearch index
-
- 'connector.document-type' = 'user', -- required: Elasticsearch document type
-
- 'update-mode' = 'append', -- optional: update mode when used as table sink.
-
- 'connector.key-delimiter' = '$', -- optional: delimiter for composite keys ("_" by default)
- -- e.g., "$" would result in IDs "KEY1$KEY2$KEY3"
-
- 'connector.key-null-literal' = 'n/a', -- optional: representation for null fields in keys ("null" by default)
-
- 'connector.failure-handler' = '...', -- optional: failure handling strategy in case a request to
- -- Elasticsearch fails ("fail" by default).
- -- valid strategies are
- -- "fail" (throws an exception if a request fails and
- -- thus causes a job failure),
- -- "ignore" (ignores failures and drops the request),
- -- "retry-rejected" (re-adds requests that have failed due
- -- to queue capacity saturation),
- -- or "custom" for failure handling with a
- -- ActionRequestFailureHandler subclass
-
- -- optional: configure how to buffer elements before sending them in bulk to the cluster for efficiency
- 'connector.flush-on-checkpoint' = 'true', -- optional: disables flushing on checkpoint (see notes below!)
- -- ("true" by default)
- 'connector.bulk-flush.max-actions' = '42', -- optional: maximum number of actions to buffer
- -- for each bulk request
- 'connector.bulk-flush.max-size' = '42 mb', -- optional: maximum size of buffered actions in bytes
- -- per bulk request
- -- (only MB granularity is supported)
- 'connector.bulk-flush.interval' = '60000', -- optional: bulk flush interval (in milliseconds)
- 'connector.bulk-flush.back-off.type' = '...', -- optional: backoff strategy ("disabled" by default)
- -- valid strategies are "disabled", "constant",
- -- or "exponential"
- 'connector.bulk-flush.back-off.max-retries' = '3', -- optional: maximum number of retries
- 'connector.bulk-flush.back-off.delay' = '30000', -- optional: delay between each backoff attempt
- -- (in milliseconds)
-
- -- optional: connection properties to be used during REST communication to Elasticsearch
- 'connector.connection-max-retry-timeout' = '3', -- optional: maximum timeout (in milliseconds)
- -- between retries
- 'connector.connection-path-prefix' = '/v1' -- optional: prefix string to be added to every
- -- REST communication
-
- 'format.type' = '...', -- required: Elasticsearch connector requires to specify a format,
- ... -- currently only 'json' format is supported.
- -- Please refer to Table Formats section for more details.
-)
-{% endhighlight %}
-</div>
</div>
**Bulk flushing:** For more information about characteristics of the optional flushing parameters see the [corresponding low-level documentation]({{ site.baseurl }}/dev/connectors/elasticsearch.html).
@@ -1091,6 +1091,38 @@ For append-only queries, the connector can also operate in [append mode](#update
The connector can be defined as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ hbase_rowkey_name rowkey_type,
+ hbase_column_family_name1 ROW<...>,
+ hbase_column_family_name2 ROW<...>
+) WITH (
+ 'connector.type' = 'hbase', -- required: specify this table type is hbase
+
+ 'connector.version' = '1.4.3', -- required: valid connector versions are "1.4.3"
+
+ 'connector.table-name' = 'hbase_table_name', -- required: hbase table name
+
+ 'connector.zookeeper.quorum' = 'localhost:2181', -- required: HBase Zookeeper quorum configuration
+ 'connector.zookeeper.znode.parent' = '/test', -- optional: the root dir in Zookeeper for HBase cluster.
+ -- The default value is "/hbase".
+
+ 'connector.write.buffer-flush.max-size' = '10mb', -- optional: writing option, determines how many size in memory of buffered
+ -- rows to insert per round trip. This can help performance on writing to JDBC
+ -- database. The default value is "2mb".
+
+ 'connector.write.buffer-flush.max-rows' = '1000', -- optional: writing option, determines how many rows to insert per round trip.
+ -- This can help performance on writing to JDBC database. No default value,
+ -- i.e. the default flushing is not depends on the number of buffered rows.
+
+ 'connector.write.buffer-flush.interval' = '2s', -- optional: writing option, sets a flush interval flushing buffered requesting
+ -- if the interval passes, in milliseconds. Default value is "0s", which means
+ -- no asynchronous flush thread will be scheduled.
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.connect(
@@ -1117,14 +1149,14 @@ The connector can be defined as follows:
connector:
type: hbase
version: "1.4.3" # required: currently only support "1.4.3"
-
+
table-name: "hbase_table_name" # required: HBase table name
-
+
zookeeper:
quorum: "localhost:2181" # required: HBase Zookeeper quorum configuration
znode.parent: "/test" # optional: the root dir in Zookeeper for HBase cluster.
# The default value is "/hbase".
-
+
write.buffer-flush:
max-size: "10mb" # optional: writing option, determines how many size in memory of buffered
# rows to insert per round trip. This can help performance on writing to JDBC
@@ -1137,41 +1169,9 @@ connector:
# no asynchronous flush thread will be scheduled.
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- hbase_rowkey_name rowkey_type,
- hbase_column_family_name1 ROW<...>,
- hbase_column_family_name2 ROW<...>
-) WITH (
- 'connector.type' = 'hbase', -- required: specify this table type is hbase
-
- 'connector.version' = '1.4.3', -- required: valid connector versions are "1.4.3"
-
- 'connector.table-name' = 'hbase_table_name', -- required: hbase table name
-
- 'connector.zookeeper.quorum' = 'localhost:2181', -- required: HBase Zookeeper quorum configuration
- 'connector.zookeeper.znode.parent' = '/test', -- optional: the root dir in Zookeeper for HBase cluster.
- -- The default value is "/hbase".
-
- 'connector.write.buffer-flush.max-size' = '10mb', -- optional: writing option, determines how many size in memory of buffered
- -- rows to insert per round trip. This can help performance on writing to JDBC
- -- database. The default value is "2mb".
-
- 'connector.write.buffer-flush.max-rows' = '1000', -- optional: writing option, determines how many rows to insert per round trip.
- -- This can help performance on writing to JDBC database. No default value,
- -- i.e. the default flushing is not depends on the number of buffered rows.
-
- 'connector.write.buffer-flush.interval' = '2s', -- optional: writing option, sets a flush interval flushing buffered requesting
- -- if the interval passes, in milliseconds. Default value is "0s", which means
- -- no asynchronous flush thread will be scheduled.
-)
-{% endhighlight %}
-</div>
</div>
-**Columns:** All the column families in HBase table must be declared as `ROW` type, the field name maps to the column family name, and the nested field names map to the column qualifier names. There is no need to declare all the families and qualifiers in the schema, users can declare what's necessary. Except the `ROW` type fields, the only one field of atomic type (e.g. `STRING`, `BIGINT`) will be recognized as row key of the table. There's no constraints on the name of row key field.
+**Columns:** All the column families in HBase table must be declared as `ROW` type, the field name maps to the column family name, and the nested field names map to the column qualifier names. There is no need to declare all the families and qualifiers in the schema, users can declare what's necessary. Except the `ROW` type fields, the only one field of atomic type (e.g. `STRING`, `BIGINT`) will be recognized as row key of the table. There's no constraints on the name of row key field.
**Temporary join:** Lookup join against HBase do not use any caching; data is always queired directly through the HBase client.
@@ -1206,69 +1206,23 @@ To use JDBC connector, need to choose an actual driver to use. Here are drivers
The connector can be defined as follows:
<div class="codetabs" markdown="1">
-<div data-lang="YAML" markdown="1">
-{% highlight yaml %}
-connector:
- type: jdbc
- url: "jdbc:mysql://localhost:3306/flink-test" # required: JDBC DB url
- table: "jdbc_table_name" # required: jdbc table name
- driver: "com.mysql.jdbc.Driver" # optional: the class name of the JDBC driver to use to connect to this URL.
- # If not set, it will automatically be derived from the URL.
-
- username: "name" # optional: jdbc user name and password
- password: "password"
-
- read: # scan options, optional, used when reading from table
- partition: # These options must all be specified if any of them is specified. In addition, partition.num must be specified. They
- # describe how to partition the table when reading in parallel from multiple tasks. partition.column must be a numeric,
- # date, or timestamp column from the table in question. Notice that lowerBound and upperBound are just used to decide
- # the partition stride, not for filtering the rows in table. So all rows in the table will be partitioned and returned.
- # This option applies only to reading.
- column: "column_name" # optional, name of the column used for partitioning the input.
- num: 50 # optional, the number of partitions.
- lower-bound: 500 # optional, the smallest value of the first partition.
- upper-bound: 1000 # optional, the largest value of the last partition.
- fetch-size: 100 # optional, Gives the reader a hint as to the number of rows that should be fetched
- # from the database when reading per round trip. If the value specified is zero, then
- # the hint is ignored. The default value is zero.
-
- lookup: # lookup options, optional, used in temporary join
- cache:
- max-rows: 5000 # optional, max number of rows of lookup cache, over this value, the oldest rows will
- # be eliminated. "cache.max-rows" and "cache.ttl" options must all be specified if any
- # of them is specified. Cache is not enabled as default.
- ttl: "10s" # optional, the max time to live for each rows in lookup cache, over this time, the oldest rows
- # will be expired. "cache.max-rows" and "cache.ttl" options must all be specified if any of
- # them is specified. Cache is not enabled as default.
- max-retries: 3 # optional, max retry times if lookup database failed
-
- write: # sink options, optional, used when writing into table
- flush:
- max-rows: 5000 # optional, flush max size (includes all append, upsert and delete records),
- # over this number of records, will flush data. The default value is "5000".
- interval: "2s" # optional, flush interval mills, over this time, asynchronous threads will flush data.
- # The default value is "0s", which means no asynchronous flush thread will be scheduled.
- max-retries: 3 # optional, max retry times if writing records to database failed.
-{% endhighlight %}
-</div>
-
<div data-lang="DDL" markdown="1">
{% highlight sql %}
CREATE TABLE MyUserTable (
...
) WITH (
'connector.type' = 'jdbc', -- required: specify this table type is jdbc
-
+
'connector.url' = 'jdbc:mysql://localhost:3306/flink-test', -- required: JDBC DB url
-
+
'connector.table' = 'jdbc_table_name', -- required: jdbc table name
-
- 'connector.driver' = 'com.mysql.jdbc.Driver', -- optional: the class name of the JDBC driver to use to connect to this URL.
+
+ 'connector.driver' = 'com.mysql.jdbc.Driver', -- optional: the class name of the JDBC driver to use to connect to this URL.
-- If not set, it will automatically be derived from the URL.
'connector.username' = 'name', -- optional: jdbc user name and password
'connector.password' = 'password',
-
+
-- scan options, optional, used when reading from table
-- These options must all be specified if any of them is specified. In addition, partition.num must be specified. They
@@ -1280,7 +1234,7 @@ CREATE TABLE MyUserTable (
'connector.read.partition.num' = '50', -- optional, the number of partitions.
'connector.read.partition.lower-bound' = '500', -- optional, the smallest value of the first partition.
'connector.read.partition.upper-bound' = '1000', -- optional, the largest value of the last partition.
-
+
'connector.read.fetch-size' = '100', -- optional, Gives the reader a hint as to the number of rows that should be fetched
-- from the database when reading per round trip. If the value specified is zero, then
-- the hint is ignored. The default value is zero.
@@ -1295,19 +1249,65 @@ CREATE TABLE MyUserTable (
'connector.lookup.max-retries' = '3', -- optional, max retry times if lookup database failed
-- sink options, optional, used when writing into table
- 'connector.write.flush.max-rows' = '5000', -- optional, flush max size (includes all append, upsert and delete records),
+ 'connector.write.flush.max-rows' = '5000', -- optional, flush max size (includes all append, upsert and delete records),
-- over this number of records, will flush data. The default value is "5000".
'connector.write.flush.interval' = '2s', -- optional, flush interval mills, over this time, asynchronous threads will flush data.
- -- The default value is "0s", which means no asynchronous flush thread will be scheduled.
+ -- The default value is "0s", which means no asynchronous flush thread will be scheduled.
'connector.write.max-retries' = '3' -- optional, max retry times if writing records to database failed
)
{% endhighlight %}
</div>
+
+<div data-lang="YAML" markdown="1">
+{% highlight yaml %}
+connector:
+ type: jdbc
+ url: "jdbc:mysql://localhost:3306/flink-test" # required: JDBC DB url
+ table: "jdbc_table_name" # required: jdbc table name
+ driver: "com.mysql.jdbc.Driver" # optional: the class name of the JDBC driver to use to connect to this URL.
+ # If not set, it will automatically be derived from the URL.
+
+ username: "name" # optional: jdbc user name and password
+ password: "password"
+
+ read: # scan options, optional, used when reading from table
+ partition: # These options must all be specified if any of them is specified. In addition, partition.num must be specified. They
+ # describe how to partition the table when reading in parallel from multiple tasks. partition.column must be a numeric,
+ # date, or timestamp column from the table in question. Notice that lowerBound and upperBound are just used to decide
+ # the partition stride, not for filtering the rows in table. So all rows in the table will be partitioned and returned.
+ # This option applies only to reading.
+ column: "column_name" # optional, name of the column used for partitioning the input.
+ num: 50 # optional, the number of partitions.
+ lower-bound: 500 # optional, the smallest value of the first partition.
+ upper-bound: 1000 # optional, the largest value of the last partition.
+ fetch-size: 100 # optional, Gives the reader a hint as to the number of rows that should be fetched
+ # from the database when reading per round trip. If the value specified is zero, then
+ # the hint is ignored. The default value is zero.
+
+ lookup: # lookup options, optional, used in temporary join
+ cache:
+ max-rows: 5000 # optional, max number of rows of lookup cache, over this value, the oldest rows will
+ # be eliminated. "cache.max-rows" and "cache.ttl" options must all be specified if any
+ # of them is specified. Cache is not enabled as default.
+ ttl: "10s" # optional, the max time to live for each rows in lookup cache, over this time, the oldest rows
+ # will be expired. "cache.max-rows" and "cache.ttl" options must all be specified if any of
+ # them is specified. Cache is not enabled as default.
+ max-retries: 3 # optional, max retry times if lookup database failed
+
+ write: # sink options, optional, used when writing into table
+ flush:
+ max-rows: 5000 # optional, flush max size (includes all append, upsert and delete records),
+ # over this number of records, will flush data. The default value is "5000".
+ interval: "2s" # optional, flush interval mills, over this time, asynchronous threads will flush data.
+ # The default value is "0s", which means no asynchronous flush thread will be scheduled.
+ max-retries: 3 # optional, max retry times if writing records to database failed.
+{% endhighlight %}
+</div>
</div>
**Upsert sink:** Flink automatically extracts valid keys from a query. For example, a query `SELECT a, b, c FROM t GROUP BY a, b` defines a composite key of the fields `a` and `b`. If a JDBC table is used as upsert sink, please make sure keys of the query is one of the unique key sets or primary key of the underlying database. This can guarantee the output result is as expected.
-**Temporary Join:** JDBC connector can be used in temporal join as a lookup source. Currently, only sync lookup mode is supported. The lookup cache options (`connector.lookup.cache.max-rows` and `connector.lookup.cache.ttl`) must all be specified if any of them is specified. The lookup cache is used to improve performance of temporal join JDBC connector by querying the cache first instead of send all requests to remote database. But the returned value might not be the latest if it is fr [...]
+**Temporary Join:** JDBC connector can be used in temporal join as a lookup source. Currently, only sync lookup mode is supported. The lookup cache options (`connector.lookup.cache.max-rows` and `connector.lookup.cache.ttl`) must all be specified if any of them is specified. The lookup cache is used to improve performance of temporal join JDBC connector by querying the cache first instead of send all requests to remote database. But the returned value might not be the latest if it is fr [...]
**Writing:** As default, the `connector.write.flush.interval` is `0s` and `connector.write.flush.max-rows` is `5000`, which means for low traffic queries, the buffered output rows may not be flushed to database for a long time. So the interval configuration is recommended to set.
@@ -1351,6 +1351,36 @@ schema is interpreted as a field renaming in the format.
The CSV format can be used as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'format.type' = 'csv', -- required: specify the schema type
+
+ 'format.fields.0.name' = 'lon', -- optional: define the schema explicitly using type information.
+ 'format.fields.0.data-type' = 'FLOAT', -- This overrides default behavior that uses table's schema as format schema.
+ 'format.fields.1.name' = 'rideTime',
+ 'format.fields.1.data-type' = 'TIMESTAMP(3)',
+
+ 'format.field-delimiter' = ';', -- optional: field delimiter character (',' by default)
+ 'format.line-delimiter' = '\r\n', -- optional: line delimiter ("\n" by default; otherwise
+ -- "\r" or "\r\n" are allowed)
+ 'format.quote-character' = '''', -- optional: quote character for enclosing field values ('"' by default)
+ 'format.allow-comments' = 'true', -- optional: ignores comment lines that start with "#"
+ -- (disabled by default);
+ -- if enabled, make sure to also ignore parse errors to allow empty rows
+ 'format.ignore-parse-errors' = 'true', -- optional: skip fields and rows with parse errors instead of failing;
+ -- fields are set to null in case of errors
+ 'format.array-element-delimiter' = '|', -- optional: the array element delimiter string for separating
+ -- array and row element values (";" by default)
+ 'format.escape-character' = '\\', -- optional: escape character for escaping values (disabled by default)
+ 'format.null-literal' = 'n/a' -- optional: null literal string that is interpreted as a
+ -- null value (disabled by default)
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.withFormat(
@@ -1427,36 +1457,6 @@ format:
# null value (disabled by default)
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'format.type' = 'csv', -- required: specify the schema type
-
- 'format.fields.0.name' = 'lon', -- optional: define the schema explicitly using type information.
- 'format.fields.0.data-type' = 'FLOAT', -- This overrides default behavior that uses table's schema as format schema.
- 'format.fields.1.name' = 'rideTime',
- 'format.fields.1.data-type' = 'TIMESTAMP(3)',
-
- 'format.field-delimiter' = ';', -- optional: field delimiter character (',' by default)
- 'format.line-delimiter' = '\r\n', -- optional: line delimiter ("\n" by default; otherwise
- -- "\r" or "\r\n" are allowed)
- 'format.quote-character' = '''', -- optional: quote character for enclosing field values ('"' by default)
- 'format.allow-comments' = 'true', -- optional: ignores comment lines that start with "#"
- -- (disabled by default);
- -- if enabled, make sure to also ignore parse errors to allow empty rows
- 'format.ignore-parse-errors' = 'true', -- optional: skip fields and rows with parse errors instead of failing;
- -- fields are set to null in case of errors
- 'format.array-element-delimiter' = '|', -- optional: the array element delimiter string for separating
- -- array and row element values (";" by default)
- 'format.escape-character' = '\\', -- optional: escape character for escaping values (disabled by default)
- 'format.null-literal' = 'n/a' -- optional: null literal string that is interpreted as a
- -- null value (disabled by default)
-)
-{% endhighlight %}
-</div>
</div>
The following table lists supported types that can be read and written:
@@ -1519,6 +1519,36 @@ If the format schema is equal to the table schema, the schema can also be automa
The JSON format can be used as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'format.type' = 'json', -- required: specify the format type
+ 'format.fail-on-missing-field' = 'true' -- optional: flag whether to fail if a field is missing or not, false by default
+
+ 'format.fields.0.name' = 'lon', -- optional: define the schema explicitly using type information.
+ 'format.fields.0.data-type' = 'FLOAT', -- This overrides default behavior that uses table's schema as format schema.
+ 'format.fields.1.name' = 'rideTime',
+ 'format.fields.1.data-type' = 'TIMESTAMP(3)',
+
+ 'format.json-schema' = -- or by using a JSON schema which parses to DECIMAL and TIMESTAMP.
+ '{ -- This also overrides the default behavior.
+ "type": "object",
+ "properties": {
+ "lon": {
+ "type": "number"
+ },
+ "rideTime": {
+ "type": "string",
+ "format": "date-time"
+ }
+ }
+ }'
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.withFormat(
@@ -1603,36 +1633,6 @@ format:
}
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'format.type' = 'json', -- required: specify the format type
- 'format.fail-on-missing-field' = 'true' -- optional: flag whether to fail if a field is missing or not, false by default
-
- 'format.fields.0.name' = 'lon', -- optional: define the schema explicitly using type information.
- 'format.fields.0.data-type' = 'FLOAT', -- This overrides default behavior that uses table's schema as format schema.
- 'format.fields.1.name' = 'rideTime',
- 'format.fields.1.data-type' = 'TIMESTAMP(3)',
-
- 'format.json-schema' = -- or by using a JSON schema which parses to DECIMAL and TIMESTAMP.
- '{ -- This also overrides the default behavior.
- "type": "object",
- "properties": {
- "lon": {
- "type": "number"
- },
- "rideTime": {
- "type": "string",
- "format": "date-time"
- }
- }
- }'
-)
-{% endhighlight %}
-</div>
</div>
The following table shows the mapping of JSON schema types to Flink SQL types:
@@ -1714,6 +1714,27 @@ The [Apache Avro](https://avro.apache.org/) format allows to read and write Avro
The Avro format can be used as follows:
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'format.type' = 'avro', -- required: specify the schema type
+ 'format.record-class' = 'org.organization.types.User', -- required: define the schema either by using an Avro specific record class
+
+ 'format.avro-schema' = -- or by using an Avro schema
+ '{
+ "type": "record",
+ "name": "test",
+ "fields" : [
+ {"name": "a", "type": "long"},
+ {"name": "b", "type": "string"}
+ ]
+ }'
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.withFormat(
@@ -1780,27 +1801,6 @@ format:
}
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'format.type' = 'avro', -- required: specify the schema type
- 'format.record-class' = 'org.organization.types.User', -- required: define the schema either by using an Avro specific record class
-
- 'format.avro-schema' = -- or by using an Avro schema
- '{
- "type": "record",
- "name": "test",
- "fields" : [
- {"name": "a", "type": "long"},
- {"name": "b", "type": "string"}
- ]
- }'
-)
-{% endhighlight %}
-</div>
</div>
Avro types are mapped to the corresponding SQL data types. Union types are only supported for specifying nullability otherwise they are converted to an `ANY` type. The following table shows the mapping:
@@ -1844,6 +1844,28 @@ replaced by a proper RFC-compliant version. Use the RFC-compliant CSV format whe
Use the old one for stream/batch filesystem operations for now.
<div class="codetabs" markdown="1">
+<div data-lang="DDL" markdown="1">
+{% highlight sql %}
+CREATE TABLE MyUserTable (
+ ...
+) WITH (
+ 'format.type' = 'csv', -- required: specify the schema type
+
+ 'format.fields.0.name' = 'lon', -- optional: declare ordered format fields explicitly. This will overrides
+ 'format.fields.0.data-type' = 'STRING', -- the default behavior that uses table's schema as format schema.
+ 'format.fields.1.name' = 'rideTime',
+ 'format.fields.1.data-type' = 'TIMESTAMP(3)',
+
+ 'format.field-delimiter' = ',', -- optional: string delimiter "," by default
+ 'format.line-delimiter' = '\n', -- optional: string delimiter "\n" by default
+ 'format.quote-character' = '"', -- optional: single character for string values, empty by default
+ 'format.comment-prefix' = '#', -- optional: string to indicate comments, empty by default
+ 'format.ignore-first-line' = 'false', -- optional: boolean flag to ignore the first line, by default it is not skipped
+ 'format.ignore-parse-errors' = 'true' -- optional: skip records with parse error instead of failing by default
+)
+{% endhighlight %}
+</div>
+
<div data-lang="Java/Scala" markdown="1">
{% highlight java %}
.withFormat(
@@ -1893,28 +1915,6 @@ format:
ignore-parse-errors: true # optional: skip records with parse error instead of failing by default
{% endhighlight %}
</div>
-
-<div data-lang="DDL" markdown="1">
-{% highlight sql %}
-CREATE TABLE MyUserTable (
- ...
-) WITH (
- 'format.type' = 'csv', -- required: specify the schema type
-
- 'format.fields.0.name' = 'lon', -- optional: declare ordered format fields explicitly. This will overrides
- 'format.fields.0.data-type' = 'STRING', -- the default behavior that uses table's schema as format schema.
- 'format.fields.1.name' = 'rideTime',
- 'format.fields.1.data-type' = 'TIMESTAMP(3)',
-
- 'format.field-delimiter' = ',', -- optional: string delimiter "," by default
- 'format.line-delimiter' = '\n', -- optional: string delimiter "\n" by default
- 'format.quote-character' = '"', -- optional: single character for string values, empty by default
- 'format.comment-prefix' = '#', -- optional: string to indicate comments, empty by default
- 'format.ignore-first-line' = 'false', -- optional: boolean flag to ignore the first line, by default it is not skipped
- 'format.ignore-parse-errors' = 'true' -- optional: skip records with parse error instead of failing by default
-)
-{% endhighlight %}
-</div>
</div>
The old CSV format is included in Flink and does not require additional dependencies.