flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-6281) Create TableSink for JDBC
Date Tue, 20 Jun 2017 13:20:00 GMT

    [ https://issues.apache.org/jira/browse/FLINK-6281?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16055738#comment-16055738
] 

ASF GitHub Bot commented on FLINK-6281:
---------------------------------------

Github user fhueske commented on a diff in the pull request:

    https://github.com/apache/flink/pull/3712#discussion_r122970495
  
    --- Diff: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCTableSink.java
---
    @@ -0,0 +1,102 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you under the Apache License, Version 2.0 (the
    + * "License"); you may not use this file except in compliance
    + * with the License.  You may obtain a copy of the License at
    + *
    + *     http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +
    +package org.apache.flink.api.java.io.jdbc;
    +
    +import org.apache.flink.api.common.functions.RuntimeContext;
    +import org.apache.flink.api.common.typeinfo.TypeInformation;
    +import org.apache.flink.api.java.typeutils.RowTypeInfo;
    +import org.apache.flink.configuration.Configuration;
    +import org.apache.flink.runtime.state.FunctionInitializationContext;
    +import org.apache.flink.runtime.state.FunctionSnapshotContext;
    +import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
    +import org.apache.flink.streaming.api.datastream.DataStream;
    +import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
    +import org.apache.flink.table.sinks.AppendStreamTableSink;
    +import org.apache.flink.table.sinks.TableSink;
    +import org.apache.flink.types.Row;
    +
    +/**
    + * An at-least-once Table sink for JDBC.
    + */
    +public class JDBCTableSink extends RichSinkFunction<Row>
    +	implements AppendStreamTableSink<Row>, CheckpointedFunction {
    +	private final JDBCOutputFormat outputFormat;
    +
    +	private String[] fieldNames;
    +	private TypeInformation[] fieldTypes;
    +
    +	public JDBCTableSink(JDBCOutputFormat outputFormat) {
    --- End diff --
    
    What do you think about not exposing the `JDBCOutputFormat` to the user, but to configure
it internally.
    
    Of course we would need many of the configuration parameters (user, pw, driver, dburl,
and table name). Users could either specify field names of the table to write to (fields are
mapped by position) or not (we use the field names of the `Table` to emit). For this information
we can construct a parameterized insert query: `INSERT INTO $table ($f1, $f2, $f3) VALUES
(?, ?, ?)`.  The field types are automatically provided by the `configure()` call.
    
    This would be a tighter integration with the Table API (using provided field types and
possibly field names).
    Does this work for your use case or do you need the flexibility of specifying your own
query?


> Create TableSink for JDBC
> -------------------------
>
>                 Key: FLINK-6281
>                 URL: https://issues.apache.org/jira/browse/FLINK-6281
>             Project: Flink
>          Issue Type: Improvement
>          Components: Table API & SQL
>            Reporter: Haohui Mai
>            Assignee: Haohui Mai
>
> It would be nice to integrate the table APIs with the JDBC connectors so that the rows
in the tables can be directly pushed into JDBC.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Mime
View raw message