How to filter kafka stream from multiples source csv files.

[amrahmed]

Hello,

I am newbie in Flink, I am stuck and looking for help,  I want to join
Streams A, B, C, D from csv source files, some of the streams update
frequently and I have another stream high throughput from Kafka K and I need
to filter K stream from [A,B,C,D].  I tried using Flink table API, Union all
streams [A,B,C,D] and then execute query SELECT * FROM K k WHERE k.key Not
IN (SELECT Key FROM ALL_UNION_LIST) but stuck in Error AppendStreamTableSink
doesn't consuming update and delete. Only select query without join get
executed and if you can let me know also How I can implement this solution
in DataStream API. All the streams have common key.

Best,
Amr



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[Guowei Ma]

Hi, Amr

What sink do you use? I think it means that the sink does not support the "upsert".

If you use Kafka as a sink[1] I think you could choose to try it after 1.12.

[1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/upsert-kafka.html

On Mon, Dec 7, 2020 at 9:19 AM amrahmed <[hidden email]> wrote:

Hello,

I am newbie in Flink, I am stuck and looking for help,  I want to join
Streams A, B, C, D from csv source files, some of the streams update
frequently and I have another stream high throughput from Kafka K and I need
to filter K stream from [A,B,C,D].  I tried using Flink table API, Union all
streams [A,B,C,D] and then execute query SELECT * FROM K k WHERE k.key Not
IN (SELECT Key FROM ALL_UNION_LIST) but stuck in Error AppendStreamTableSink
doesn't consuming update and delete. Only select query without join get
executed and if you can let me know also How I can implement this solution
in DataStream API. All the streams have common key.

Best,
Amr



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/