Repository: flink Updated Branches: refs/heads/master 1de8acc3c -> f78eb0f8b
[hotfix] [docs] Improve TableSink documentation. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f78eb0f8 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f78eb0f8 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f78eb0f8 Branch: refs/heads/master Commit: f78eb0f8bb2912a6eb874ad1a89a10f78c3a477a Parents: 43e5a81 Author: Fabian Hueske <fhue...@apache.org> Authored: Fri Aug 11 11:33:26 2017 +0200 Committer: Fabian Hueske <fhue...@apache.org> Committed: Fri Aug 11 17:29:20 2017 +0200 ---------------------------------------------------------------------- docs/dev/table/sourceSinks.md | 70 ++++++++++++++++++++++++++++++++++---- 1 file changed, 64 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/f78eb0f8/docs/dev/table/sourceSinks.md ---------------------------------------------------------------------- diff --git a/docs/dev/table/sourceSinks.md b/docs/dev/table/sourceSinks.md index 53b93e1..7163c67 100644 --- a/docs/dev/table/sourceSinks.md +++ b/docs/dev/table/sourceSinks.md @@ -34,8 +34,6 @@ Have a look at the [common concepts and API](common.html) page for details how t Provided TableSources --------------------- -**TODO: extend and complete** - Currently, Flink provides the `CsvTableSource` to read CSV files and a few table sources to read JSON or Avro data from Kafka. A custom `TableSource` can be defined by implementing the `BatchTableSource` or `StreamTableSource` interface. See section on [defining a custom TableSource](#define-a-tablesource) for details. @@ -202,9 +200,63 @@ val csvTableSource = CsvTableSource Provided TableSinks ------------------- -### JDBCAppendSink +The following table lists the `TableSink`s which are provided with Flink. + +| **Class name** | **Maven dependency** | **Batch?** | **Streaming?** | **Description** +| `CsvTableSink` | `flink-table` | Y | Append | A simple sink for CSV files. +| `JDBCAppendTableSink` | `flink-jdbc` | Y | Append | Writes tables to a JDBC database. +| `Kafka08JsonTableSink` | `flink-connector-kafka-0.8` | N | Append | A Kafka 0.8 sink with JSON encoding. +| `Kafka09JsonTableSink` | `flink-connector-kafka-0.9` | N | Append | A Kafka 0.9 sink with JSON encoding. + +All sinks that come with the `flink-table` dependency can be directly used by your Table programs. For all other table sinks, you have to add the respective dependency in addition to the `flink-table` dependency. + +A custom `TableSink` can be defined by implementing the `BatchTableSink`, `AppendStreamTableSink`, `RetractStreamTableSink`, or `UpsertStreamTableSink` interface. See section on [defining a custom TableSink](#define-a-tablesink) for details. + +{% top %} + +### CsvTableSink + +The `CsvTableSink` emits a `Table` to one or more CSV files. + +The sink only supports append-only streaming tables. It cannot be used to emit a `Table` that is continuously updated. See the [documentation on Table to Stream conversions](./streaming.html#table-to-stream-conversion) for details. When emitting a streaming table, rows are written at least once (if checkpointing is enabled) and the `CsvTableSink` does not split output files into bucket files but continuously writes to the same files. + +<div class="codetabs" markdown="1"> +<div data-lang="java" markdown="1"> +{% highlight java %} + +Table table = ... + +table.writeToSink( + new CsvTableSink( + path, // output path + "|", // optional: delimit files by '|' + 1, // optional: write to a single file + WriteMode.OVERWRITE)); // optional: override existing files -<code>JDBCAppendSink</code> allows you to bridge the data stream to the JDBC driver. The sink only supports append-only data. It does not support retractions and upserts from Flink's perspectives. However, you can customize the query using <code>REPLACE</code> or <code>INSERT OVERWRITE</code> to implement upsert inside the database. +{% endhighlight %} +</div> + +<div data-lang="scala" markdown="1"> +{% highlight scala %} + +val table: Table = ??? + +table.writeToSink( + new CsvTableSink( + path, // output path + fieldDelim = "|", // optional: delimit files by '|' + numFiles = 1, // optional: write to a single file + writeMode = WriteMode.OVERWRITE)) // optional: override existing files + +{% endhighlight %} +</div> +</div> + +### JDBCAppendTableSink + +The `JDBCAppendTableSink` emits a `Table` to a JDBC connection. The sink only supports append-only streaming tables. It cannot be used to emit a `Table` that is continuously updated. See the [documentation on Table to Stream conversions](./streaming.html#table-to-stream-conversion) for details. + +The `JDBCAppendTableSink` inserts each `Table` row at least once into the database table (if checkpointing is enabled). However, you can specify the insertion query using <code>REPLACE</code> or <code>INSERT OVERWRITE</code> to perform upsert writes to the database. To use the JDBC sink, you have to add the JDBC connector dependency (<code>flink-jdbc</code>) to your project. Then you can create the sink using <code>JDBCAppendSinkBuilder</code>: @@ -218,22 +270,28 @@ JDBCAppendTableSink sink = JDBCAppendTableSink.builder() .setQuery("INSERT INTO books (id) VALUES (?)") .setParameterTypes(INT_TYPE_INFO) .build(); + +Table table = ... +table.writeToSink(sink); {% endhighlight %} </div> <div data-lang="scala" markdown="1"> {% highlight scala %} -val sink = JDBCAppendTableSink.builder() +val sink: JDBCAppendTableSink = JDBCAppendTableSink.builder() .setDrivername("org.apache.derby.jdbc.EmbeddedDriver") .setDBUrl("jdbc:derby:memory:ebookshop") .setQuery("INSERT INTO books (id) VALUES (?)") .setParameterTypes(INT_TYPE_INFO) .build() + +val table: Table = ??? +table.writeToSink(sink) {% endhighlight %} </div> </div> -Similar to using <code>JDBCOutputFormat</code>, you have to explicitly specify the name of the JDBC driver, the JDBC URL, the query to be executed, and the field types of the JDBC table. You can connect the sink with other <code>DataStream</code>s once the sink is constructed. +Similar to using <code>JDBCOutputFormat</code>, you have to explicitly specify the name of the JDBC driver, the JDBC URL, the query to be executed, and the field types of the JDBC table. {% top %}
