(DEPRECATED) Apache Flink User Mailing List archive.

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

Classic

List

Threaded

8 messages Options

wanglei2@geekplus.com.cn

Mar 25, 2020; 3:12am

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

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]

Jingsong Li

Mar 25, 2020; 3:14am

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

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]

... [show rest of quote]

Best, Jingsong Lee

wanglei2@geekplus.com.cn

Mar 25, 2020; 3:25am

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

Thanks Jingsong.

When executing this sql, the mysql table record can be deleted. So i guess it is a retract stream.

I want to know the exactly java code it is generated and have a look at it.

Thanks,

Lei

[hidden email]

Sender: [hidden email]
Send Time: 2020-03-25 11:14
Receiver: [hidden email]
cc: [hidden email]
Subject: Re: Where can i find MySQL retract stream table sink java soure code？
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]
... [show rest of quote]
--
Best, Jingsong Lee

... [show rest of quote]

wanglei2@geekplus.com.cn

Mar 25, 2020; 3:32am

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

Seems it is here: https://github.com/apache/flink/tree/master/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc

There's no JDBCRetractTableSink, only append and upsert.

I am confused why the MySQL record can be deleted.

Thanks,

Lei

[hidden email]

Sender: [hidden email]
Send Time: 2020-03-25 11:25
Receiver: [hidden email]
cc: [hidden email]
Subject: Re: Re: Where can i find MySQL retract stream table sink java soure code？
Thanks Jingsong.

When executing this sql, the mysql table record can be deleted. So i guess it is a retract stream.
I want to know the exactly java code it is generated and have a look at it.

Thanks,
Lei

[hidden email]
Sender: [hidden email]
Send Time: 2020-03-25 11:14
Receiver: [hidden email]
cc: [hidden email]
Subject: Re: Where can i find MySQL retract stream table sink java soure code？
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]
... [show rest of quote]
--
Best, Jingsong Lee
... [show rest of quote]

... [show rest of quote]

Jingsong Li

Mar 25, 2020; 3:37am

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

In reply to this post by Jingsong Li

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]
... [show rest of quote]
--
Best, Jingsong Lee

... [show rest of quote]

Best, Jingsong Lee

Jingsong Li

Mar 25, 2020; 3:39am

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]
... [show rest of quote]
--
Best, Jingsong Lee
... [show rest of quote]
--
Best, Jingsong Lee

... [show rest of quote]

Best, Jingsong Lee

wanglei2@geekplus.com.cn

Mar 25, 2020; 3:57am

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

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]
... [show rest of quote]
--
Best, Jingsong Lee
... [show rest of quote]
--
Best, Jingsong Lee
... [show rest of quote]
--
Best, Jingsong Lee

... [show rest of quote]

Jingsong Li

Mar 25, 2020; 4:29am

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

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]
... [show rest of quote]
--
Best, Jingsong Lee
... [show rest of quote]
--
Best, Jingsong Lee
... [show rest of quote]
--
Best, Jingsong Lee
... [show rest of quote]

... [show rest of quote]

Best, Jingsong Lee