You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by elakiya udhayanan <la...@gmail.com> on 2023/12/06 09:49:17 UTC
Query on using two sinks for a Flink job (Flink SQL)
Hi Team,
I would like to know the possibility of having two sinks in a single Flink
job. In my case I am using the Flink SQL based job where I try to consume
from two different Kafka topics using the create table (as below) DDL and
then use a join condition to correlate them and at present write it to an
external database (PostgreSQL - as a sink). I would like to know if I can
add another sink where I want to also write it to kafka topic (as the
second sink).
I tried using two sql scripts (two create and two insert for the same) but
was facing an exception* "Cannot have more than one execute() or
executeAsync() call in a single environment. at "*
Also tried to use the StatementSet functionality which again gave me an
exception *"org.apache.flink.table.api.TableException: Only insert
statement is supported now. at ".*
I am looking for some help in regards to this. TIA
*Note:* I am using the Flink UI to submit my job.
*Sample DDL statement used:*String statement = "CREATE TABLE Person (\r\n" +
" person ROW(id STRING, name STRING\r\n" +
" ),\r\n" +
" PRIMARY KEY (id) NOT ENFORCED\r\n" +
") WITH (\r\n" +
" 'connector' = 'upsert-kafka',\r\n" +
" 'topic' = 'employee',\r\n" +
" 'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',\r\n" +
" 'key.format' = 'raw',\r\n" +
" 'value.format' = 'avro-confluent',\r\n" +
" 'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n"
+
")";
Thanks,
Elakiya
Re: Query on using two sinks for a Flink job (Flink SQL)
Posted by Feng Jin <ji...@gmail.com>.
Hi Elakiya,
You should use DML in the statement set instead of DQL .
Here is a simple example:
executeSql("CREATE TABLE source_table1 ......");
executeSql("CREATE TABLE source_table2 ......");
executeSql("CREATE TABLE sink_table1 ......");
executeSql("CREATE TABLE sink_table1 ......");
stmtSet.addInsertSql("INSERT INTO sink_tabl1 SELECT xxx from source_table1
join source_table2 ...");
stmtSet.addInsertSql("INSERT INTO sink_tabl2 SELECT xxx from source_table1
join source_table2 ...");
stmtSet.execute();
Best,
Feng
On Thu, Dec 7, 2023 at 12:48 AM elakiya udhayanan <la...@gmail.com>
wrote:
> Hi Xuyang, Zhangao,
>
> Thanks for your response, I have attached sample job files that I tried
> with the Statementset and with two queries. Please let me know if you are
> able to point out where I am possibly going wrong.
>
> Thanks,
> Elakiya
>
> On Wed, Dec 6, 2023 at 4:51 PM Xuyang <xy...@163.com> wrote:
>
>> Hi, Elakiya.
>> Are you following the example here[1]? Could you attach a minimal,
>> reproducible SQL?
>>
>> [1]
>> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/insert/
>>
>>
>>
>> --
>> Best！
>> Xuyang
>>
>>
>> At 2023-12-06 17:49:17, "elakiya udhayanan" <la...@gmail.com> wrote:
>>
>> Hi Team,
>> I would like to know the possibility of having two sinks in a
>> single Flink job. In my case I am using the Flink SQL based job where I try
>> to consume from two different Kafka topics using the create table (as
>> below) DDL and then use a join condition to correlate them and at present
>> write it to an external database (PostgreSQL - as a sink). I would like to
>> know if I can add another sink where I want to also write it to kafka topic
>> (as the second sink).
>> I tried using two sql scripts (two create and two insert for the same)
>> but was facing an exception* "Cannot have more than one execute() or
>> executeAsync() call in a single environment. at "*
>> Also tried to use the StatementSet functionality which again gave me an
>> exception *"org.apache.flink.table.api.TableException: Only insert
>> statement is supported now. at ".*
>> I am looking for some help in regards to this. TIA
>>
>> *Note:* I am using the Flink UI to submit my job.
>>
>> *Sample DDL statement used:*String statement = "CREATE TABLE Person
>> (\r\n" +
>> " person ROW(id STRING, name STRING\r\n" +
>> " ),\r\n" +
>> " PRIMARY KEY (id) NOT ENFORCED\r\n" +
>> ") WITH (\r\n" +
>> " 'connector' = 'upsert-kafka',\r\n" +
>> " 'topic' = 'employee',\r\n" +
>> " 'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',\r\n" +
>> " 'key.format' = 'raw',\r\n" +
>> " 'value.format' = 'avro-confluent',\r\n" +
>> " 'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n"
>> +
>> ")";
>>
>> Thanks,
>> Elakiya
>>
>>
Re: Query on using two sinks for a Flink job (Flink SQL)
Posted by elakiya udhayanan <la...@gmail.com>.
Hi Xuyang, Zhangao,
Thanks for your response, I have attached sample job files that I tried
with the Statementset and with two queries. Please let me know if you are
able to point out where I am possibly going wrong.
Thanks,
Elakiya
On Wed, Dec 6, 2023 at 4:51 PM Xuyang <xy...@163.com> wrote:
> Hi, Elakiya.
> Are you following the example here[1]? Could you attach a minimal,
> reproducible SQL?
>
> [1]
> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/insert/
>
>
>
> --
> Best！
> Xuyang
>
>
> At 2023-12-06 17:49:17, "elakiya udhayanan" <la...@gmail.com> wrote:
>
> Hi Team,
> I would like to know the possibility of having two sinks in a
> single Flink job. In my case I am using the Flink SQL based job where I try
> to consume from two different Kafka topics using the create table (as
> below) DDL and then use a join condition to correlate them and at present
> write it to an external database (PostgreSQL - as a sink). I would like to
> know if I can add another sink where I want to also write it to kafka topic
> (as the second sink).
> I tried using two sql scripts (two create and two insert for the same) but
> was facing an exception* "Cannot have more than one execute() or
> executeAsync() call in a single environment. at "*
> Also tried to use the StatementSet functionality which again gave me an
> exception *"org.apache.flink.table.api.TableException: Only insert
> statement is supported now. at ".*
> I am looking for some help in regards to this. TIA
>
> *Note:* I am using the Flink UI to submit my job.
>
> *Sample DDL statement used:*String statement = "CREATE TABLE Person
> (\r\n" +
> " person ROW(id STRING, name STRING\r\n" +
> " ),\r\n" +
> " PRIMARY KEY (id) NOT ENFORCED\r\n" +
> ") WITH (\r\n" +
> " 'connector' = 'upsert-kafka',\r\n" +
> " 'topic' = 'employee',\r\n" +
> " 'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',\r\n" +
> " 'key.format' = 'raw',\r\n" +
> " 'value.format' = 'avro-confluent',\r\n" +
> " 'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n"
> +
> ")";
>
> Thanks,
> Elakiya
>
>
Re:Query on using two sinks for a Flink job (Flink SQL)
Posted by Xuyang <xy...@163.com>.
Hi, Elakiya.
Are you following the example here[1]? Could you attach a minimal, reproducible SQL?
[1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/insert/
--
Best！
Xuyang
At 2023-12-06 17:49:17, "elakiya udhayanan" <la...@gmail.com> wrote:
Hi Team,
I would like to know the possibility of having two sinks in a single Flink job. In my case I am using the Flink SQL based job where I try to consume from two different Kafka topics using the create table (as below) DDL and then use a join condition to correlate them and at present write it to an external database (PostgreSQL - as a sink). I would like to know if I can add another sink where I want to also write it to kafka topic (as the second sink).
I tried using two sql scripts (two create and two insert for the same) but was facing an exception "Cannot have more than one execute() or executeAsync() call in a single environment. at "
Also tried to use the StatementSet functionality which again gave me an exception "org.apache.flink.table.api.TableException: Only insert statement is supported now. at ".
I am looking for some help in regards to this. TIA
Note: I am using the Flink UI to submit my job.
Sample DDL statement used:
String statement = "CREATE TABLE Person (\r\n" +
" person ROW(id STRING, name STRING\r\n" +
" ),\r\n" +
" PRIMARY KEY (id) NOT ENFORCED\r\n" +
") WITH (\r\n" +
" 'connector' = 'upsert-kafka',\r\n" +
" 'topic' = 'employee',\r\n" +
" 'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',\r\n" +
" 'key.format' = 'raw',\r\n" +
" 'value.format' = 'avro-confluent',\r\n" +
" 'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n" +
")";
Thanks,
Elakiya
Re: Query on using two sinks for a Flink job (Flink SQL)
Posted by elakiya udhayanan <la...@gmail.com>.
Hi Chen/ Feng,
Thanks for pointing out the mistake I made, after correcting the query I am
able to run the job with two sinks successfully.
Thanks,
Elakiya
On Thu, Dec 7, 2023 at 4:37 AM Chen Yu <yu...@gmail.com> wrote:
> Hi Chen,
> You should tell flink which table to insert by “INSERT INTO XXX SELECT
> XXX”.
>
> For single non insert query, flink will collect output to the console
> automatically. Therefore, you don’t need to add insert also works.
>
> But you must point out target table specifically when you need to write
> data to external storage.
>
> Like,
>
> String relateQuery = "insert into xxx select correlator_id , name, relationship from Correlation; ;
>
>
> Best,
> Yu Chen
>
>
> 获取 Outlook for iOS <https://aka.ms/o0ukef>
> ------------------------------
> *发件人:* Zhanghao Chen <zh...@outlook.com>
> *发送时间:* Wednesday, December 6, 2023 7:21:50 PM
> *收件人:* elakiya udhayanan <la...@gmail.com>; user@flink.apache.org <
> user@flink.apache.org>
> *主题:* Re: Query on using two sinks for a Flink job (Flink SQL)
>
> Hi Elakiya,
>
> You can try executing TableEnvironmentImpl#executeInternal for non-insert
> statements, then using StatementSet.addInsertSql to add multiple insertion
> statetments, and finally calling StatementSet#execute.
>
> Best,
> Zhanghao Chen
> ------------------------------
> *From:* elakiya udhayanan <la...@gmail.com>
> *Sent:* Wednesday, December 6, 2023 17:49
> *To:* user@flink.apache.org <us...@flink.apache.org>
> *Subject:* Query on using two sinks for a Flink job (Flink SQL)
>
> Hi Team,
> I would like to know the possibility of having two sinks in a
> single Flink job. In my case I am using the Flink SQL based job where I try
> to consume from two different Kafka topics using the create table (as
> below) DDL and then use a join condition to correlate them and at present
> write it to an external database (PostgreSQL - as a sink). I would like to
> know if I can add another sink where I want to also write it to kafka topic
> (as the second sink).
> I tried using two sql scripts (two create and two insert for the same) but
> was facing an exception* "Cannot have more than one execute() or
> executeAsync() call in a single environment. at "*
> Also tried to use the StatementSet functionality which again gave me an
> exception *"org.apache.flink.table.api.TableException: Only insert
> statement is supported now. at ".*
> I am looking for some help in regards to this. TIA
>
> *Note:* I am using the Flink UI to submit my job.
>
> *Sample DDL statement used: *String statement = "CREATE TABLE Person
> (\r\n" +
> " person ROW(id STRING, name STRING\r\n" +
> " ),\r\n" +
> " PRIMARY KEY (id) NOT ENFORCED\r\n" +
> ") WITH (\r\n" +
> " 'connector' = 'upsert-kafka',\r\n" +
> " 'topic' = 'employee',\r\n" +
> " 'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',\r\n" +
> " 'key.format' = 'raw',\r\n" +
> " 'value.format' = 'avro-confluent',\r\n" +
> " 'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n"
> +
> ")";
>
> Thanks,
> Elakiya
>
Re: Query on using two sinks for a Flink job (Flink SQL)
Posted by Chen Yu <yu...@gmail.com>.
Hi Chen,
You should tell flink which table to insert by “INSERT INTO XXX SELECT XXX”.
For single non insert query, flink will collect output to the console automatically. Therefore, you don’t need to add insert also works.
But you must point out target table specifically when you need to write data to external storage.
Like,
String relateQuery = "insert into xxx select correlator_id , name, relationship from Correlation; ;
Best,
Yu Chen
获取 Outlook for iOS<https://aka.ms/o0ukef>
________________________________
发件人: Zhanghao Chen <zh...@outlook.com>
发送时间: Wednesday, December 6, 2023 7:21:50 PM
收件人: elakiya udhayanan <la...@gmail.com>; user@flink.apache.org <us...@flink.apache.org>
主题: Re: Query on using two sinks for a Flink job (Flink SQL)
Hi Elakiya,
You can try executing TableEnvironmentImpl#executeInternal for non-insert statements, then using StatementSet.addInsertSql to add multiple insertion statetments, and finally calling StatementSet#execute.
Best,
Zhanghao Chen
________________________________
From: elakiya udhayanan <la...@gmail.com>
Sent: Wednesday, December 6, 2023 17:49
To: user@flink.apache.org <us...@flink.apache.org>
Subject: Query on using two sinks for a Flink job (Flink SQL)
Hi Team,
I would like to know the possibility of having two sinks in a single Flink job. In my case I am using the Flink SQL based job where I try to consume from two different Kafka topics using the create table (as below) DDL and then use a join condition to correlate them and at present write it to an external database (PostgreSQL - as a sink). I would like to know if I can add another sink where I want to also write it to kafka topic (as the second sink).
I tried using two sql scripts (two create and two insert for the same) but was facing an exception "Cannot have more than one execute() or executeAsync() call in a single environment. at "
Also tried to use the StatementSet functionality which again gave me an exception "org.apache.flink.table.api.TableException: Only insert statement is supported now. at ".
I am looking for some help in regards to this. TIA
Note: I am using the Flink UI to submit my job.
Sample DDL statement used:
String statement = "CREATE TABLE Person (\r\n" +
" person ROW(id STRING, name STRING\r\n" +
" ),\r\n" +
" PRIMARY KEY (id) NOT ENFORCED\r\n" +
") WITH (\r\n" +
" 'connector' = 'upsert-kafka',\r\n" +
" 'topic' = 'employee',\r\n" +
" 'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',\r\n" +
" 'key.format' = 'raw',\r\n" +
" 'value.format' = 'avro-confluent',\r\n" +
" 'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n" +
")";
Thanks,
Elakiya
Re: Query on using two sinks for a Flink job (Flink SQL)
Posted by Zhanghao Chen <zh...@outlook.com>.
Hi Elakiya,
You can try executing TableEnvironmentImpl#executeInternal for non-insert statements, then using StatementSet.addInsertSql to add multiple insertion statetments, and finally calling StatementSet#execute.
Best,
Zhanghao Chen
________________________________
From: elakiya udhayanan <la...@gmail.com>
Sent: Wednesday, December 6, 2023 17:49
To: user@flink.apache.org <us...@flink.apache.org>
Subject: Query on using two sinks for a Flink job (Flink SQL)
Hi Team,
I would like to know the possibility of having two sinks in a single Flink job. In my case I am using the Flink SQL based job where I try to consume from two different Kafka topics using the create table (as below) DDL and then use a join condition to correlate them and at present write it to an external database (PostgreSQL - as a sink). I would like to know if I can add another sink where I want to also write it to kafka topic (as the second sink).
I tried using two sql scripts (two create and two insert for the same) but was facing an exception "Cannot have more than one execute() or executeAsync() call in a single environment. at "
Also tried to use the StatementSet functionality which again gave me an exception "org.apache.flink.table.api.TableException: Only insert statement is supported now. at ".
I am looking for some help in regards to this. TIA
Note: I am using the Flink UI to submit my job.
Sample DDL statement used:
String statement = "CREATE TABLE Person (\r\n" +
" person ROW(id STRING, name STRING\r\n" +
" ),\r\n" +
" PRIMARY KEY (id) NOT ENFORCED\r\n" +
") WITH (\r\n" +
" 'connector' = 'upsert-kafka',\r\n" +
" 'topic' = 'employee',\r\n" +
" 'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',\r\n" +
" 'key.format' = 'raw',\r\n" +
" 'value.format' = 'avro-confluent',\r\n" +
" 'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n" +
")";
Thanks,
Elakiya