Re: Re: Where can i find MySQL retract stream table sink java soure code？

Posted by Jingsong Li on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Where-can-i-find-MySQL-retract-stream-table-sink-java-soure-code-tp33888p33895.html

It is not the mean what you said.

There are two queries: append query and update query.

For update query, there are two ways to handle, one is retract, another is upsert.

So the thing is a sink can choose a mode to handle update query. Just choose one is OK.

You could read more in [1].

[1]https://ci.apache.org/projects/flink/flink-docs-master/dev/table/streaming/dynamic_tables.html

Best,

Jingsong Lee

On Wed, Mar 25, 2020 at 11:57 AM [hidden email] <[hidden email]> wrote:

Thanks Jingsong.

So JDBCTableSink now suport append and upsert mode.  Retract mode not  available yet. It is right？

Thanks,

Lei 

---

[hidden email] 

 

Sender: [hidden email]

Send Time: 2020-03-25 11:39

Receiver: [hidden email]

cc: [hidden email]

Subject: Re: Where can i find MySQL retract stream table sink java soure code？

Hi,

Maybe you have some misunderstanding to upsert sink. You can take a look to [1], it can deal with "delete" records.

[1]https://ci.apache.org/projects/flink/flink-docs-master/dev/table/streaming/dynamic_tables.html#table-to-stream-conversion

Best,

Jingsong Lee

On Wed, Mar 25, 2020 at 11:37 AM Jingsong Li <[hidden email]> wrote:

Hi,

This can be a upsert stream [1], and JDBC has upsert sink now [2].

[1]https://ci.apache.org/projects/flink/flink-docs-master/dev/table/streaming/dynamic_tables.html#table-to-stream-conversion

[2]https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html#jdbc-connector

Best,

Jingsong Lee

On Wed, Mar 25, 2020 at 11:14 AM Jingsong Li <[hidden email]> wrote:

Hi,

This can be a upsert stream [1]

Best,

Jingsong Lee

On Wed, Mar 25, 2020 at 11:12 AM [hidden email] <[hidden email]> wrote:

Create one table with kafka,  another table with MySQL  using flinksql.

Write a sql to read from kafka and write to MySQL.

INSERT INTO mysqlTable  SELECT status, COUNT(order_no) AS num FROM 
                        (SELECT order_no, LAST_VALUE(status) AS status FROM kafkaTable GROUP BY order_no) 
                        GROUP BY status

I think this is a retract stream. 

But where can i find the java source code  about MySQL retract table sink?

Thanks,

Lei

---

[hidden email] 

--

Best, Jingsong Lee

--

Best, Jingsong Lee

--

Best, Jingsong Lee

--