(DEPRECATED) Apache Flink User Mailing List archive.

Flink 1.10 es sink exception

classic Classic list List threaded Threaded
♦
♦
 7 messages Options
sunfulin
Reply | Threaded
Open this post in threaded view
♦
♦
|

Flink 1.10 es sink exception

sunfulin
Hi, guys
When running the same Flink sql like the following, I met exception like "org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated". I am using the latest Flink 1.10 release with blink planner enabled. Because the same logic runs well within Flink 1.8.2 old planner. Does the SQL usage has some problem or may has a bug here ? 


INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clickCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min



 
sunfulin
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re:Flink 1.10 es sink exception

sunfulin
Anyone can share a little advice on the reason of this exception? I changed to use old planner, the same sql runs well. 





At 2020-02-13 16:07:18, "sunfulin" <[hidden email]> wrote:

Hi, guys
When running the same Flink sql like the following, I met exception like "org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated". I am using the latest Flink 1.10 release with blink planner enabled. Because the same logic runs well within Flink 1.8.2 old planner. Does the SQL usage has some problem or may has a bug here ? 


INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clickCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min



 



 
Jark Wu-3
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Flink 1.10 es sink exception

Jark Wu-3
Hi sunfulin,

Is this the real query you submit?  AFAIK, insert with column list is not allowed for now, 
i.e. the `INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)`.

Could you attach the full SQL text, including  DDLs of ES6_ZHANGLE_OUTPUT table and kafka_zl_etrack_event_stream table.
If you have a minimal program that can reproduce this problem, that would be great. 

Best,
Jark

On Fri, 14 Feb 2020 at 22:53, Robert Metzger <[hidden email]> wrote:


---------- Forwarded message ---------
From: sunfulin <[hidden email]>
Date: Fri, Feb 14, 2020 at 2:59 AM
Subject: Re:Flink 1.10 es sink exception
To: [hidden email] <[hidden email]>


Anyone can share a little advice on the reason of this exception? I changed to use old planner, the same sql runs well. 





At 2020-02-13 16:07:18, "sunfulin" <[hidden email]> wrote:

Hi, guys
When running the same Flink sql like the following, I met exception like "org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated". I am using the latest Flink 1.10 release with blink planner enabled. Because the same logic runs well within Flink 1.8.2 old planner. Does the SQL usage has some problem or may has a bug here ? 


INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clickCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min



 



 
sunfulin
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re:Re: Flink 1.10 es sink exception

sunfulin
Hi, Jark
Appreciate for your reply. insert with column list indeed is not allowed with old planner enabled in Flink 1.10 while it will throws exception such as "Partial insert is not supported". 
Never mind for this issue. Focus on the UpsertMode exception, my es DDL is like the following: 

CREATE TABLE ES6_ZHANGLE_OUTPUT (
  aggId varchar ,
  pageId varchar ,
  ts varchar ,
  expoCnt bigint ,
  clkCnt bigint
) WITH (
'connector.type' = 'elasticsearch',
'connector.version' = '6',
'connector.hosts' = 'http://168.61.113.171:9092;http://168.61.113.171:9093',
'connector.index' = 'flink_zhangle_pageview',
'connector.document-type' = '_doc',
'update-mode' = 'upsert',
'connector.key-delimiter' = '$',
'connector.key-null-literal' = 'n/a',
'connector.bulk-flush.interval' = '1000',
'format.type' = 'json'
)


And the SQL logic is as the following:

INSERT INTO ES6_ZHANGLE_OUTPUT
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clkCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min

I simply run StreamTableEnvironment.sqlUpdate( the above sql content) and execute the task. Not sure what the root cause is. 


At 2020-02-14 23:19:14, "Jark Wu" <[hidden email]> wrote:

Hi sunfulin,

Is this the real query you submit?  AFAIK, insert with column list is not allowed for now, 
i.e. the `INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)`.

Could you attach the full SQL text, including  DDLs of ES6_ZHANGLE_OUTPUT table and kafka_zl_etrack_event_stream table.
If you have a minimal program that can reproduce this problem, that would be great. 

Best,
Jark

On Fri, 14 Feb 2020 at 22:53, Robert Metzger <[hidden email]> wrote:


---------- Forwarded message ---------
From: sunfulin <[hidden email]>
Date: Fri, Feb 14, 2020 at 2:59 AM
Subject: Re:Flink 1.10 es sink exception
To: [hidden email] <[hidden email]>


Anyone can share a little advice on the reason of this exception? I changed to use old planner, the same sql runs well. 





At 2020-02-13 16:07:18, "sunfulin" <[hidden email]> wrote:

Hi, guys
When running the same Flink sql like the following, I met exception like "org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated". I am using the latest Flink 1.10 release with blink planner enabled. Because the same logic runs well within Flink 1.8.2 old planner. Does the SQL usage has some problem or may has a bug here ? 


INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clickCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min



 



 



 
Leonard Xu
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Flink 1.10 es sink exception

Leonard Xu

Hi, sunfulin
I reproduce your case，this should be a bug in extracting unique key from plan and I create an issue[1] to trace this.


CC:  jark


[1]https://issues.apache.org/jira/browse/FLINK-16070



在 2020年2月14日，23:39，sunfulin <[hidden email]> 写道：

Hi, Jark
Appreciate for your reply. insert with column list indeed is not allowed with old planner enabled in Flink 1.10 while it will throws exception such as "Partial insert is not supported". 
Never mind for this issue. Focus on the UpsertMode exception, my es DDL is like the following: 

CREATE TABLE ES6_ZHANGLE_OUTPUT (
  aggId varchar ,
  pageId varchar ,
  ts varchar ,
  expoCnt bigint ,
  clkCnt bigint
) WITH (
'connector.type' = 'elasticsearch',
'connector.version' = '6',
'connector.hosts' = '<a href="http://168.61.113.171:9092;http://168.61.113.171:9093" class="">http://168.61.113.171:9092;http://168.61.113.171:9093',
'connector.index' = 'flink_zhangle_pageview',
'connector.document-type' = '_doc',
'update-mode' = 'upsert',
'connector.key-delimiter' = '$',
'connector.key-null-literal' = 'n/a',
'connector.bulk-flush.interval' = '1000',
'format.type' = 'json'
)


And the SQL logic is as the following:

INSERT INTO ES6_ZHANGLE_OUTPUT
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clkCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min

I simply run StreamTableEnvironment.sqlUpdate( the above sql content) and execute the task. Not sure what the root cause is. 


At 2020-02-14 23:19:14, "Jark Wu" <[hidden email]> wrote:

Hi sunfulin,

Is this the real query you submit?  AFAIK, insert with column list is not allowed for now, 
i.e. the `INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)`.

Could you attach the full SQL text, including  DDLs of ES6_ZHANGLE_OUTPUT table and kafka_zl_etrack_event_stream table.
If you have a minimal program that can reproduce this problem, that would be great. 

Best,
Jark

On Fri, 14 Feb 2020 at 22:53, Robert Metzger <[hidden email]> wrote:


---------- Forwarded message ---------
From: sunfulin <[hidden email]>
Date: Fri, Feb 14, 2020 at 2:59 AM
Subject: Re:Flink 1.10 es sink exception
To: [hidden email] <[hidden email]>


Anyone can share a little advice on the reason of this exception? I changed to use old planner, the same sql runs well. 





At 2020-02-13 16:07:18, "sunfulin" <[hidden email]> wrote:

Hi, guys
When running the same Flink sql like the following, I met exception like "org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated". I am using the latest Flink 1.10 release with blink planner enabled. Because the same logic runs well within Flink 1.8.2 old planner. Does the SQL usage has some problem or may has a bug here ? 


INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clickCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min



 


 


 
sunfulin
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re:Re: Flink 1.10 es sink exception

sunfulin
Hi，
WOW，really thankful for the track and debug of this problem. I can see the constant key issue. Appreciate for your kindly help : )





At 2020-02-15 21:06:58, "Leonard Xu" <[hidden email]> wrote:


Hi, sunfulin
I reproduce your case，this should be a bug in extracting unique key from plan and I create an issue[1] to trace this.


CC:  jark


[1]https://issues.apache.org/jira/browse/FLINK-16070



在 2020年2月14日，23:39，sunfulin <[hidden email]> 写道：

Hi, Jark
Appreciate for your reply. insert with column list indeed is not allowed with old planner enabled in Flink 1.10 while it will throws exception such as "Partial insert is not supported". 
Never mind for this issue. Focus on the UpsertMode exception, my es DDL is like the following: 

CREATE TABLE ES6_ZHANGLE_OUTPUT (
  aggId varchar ,
  pageId varchar ,
  ts varchar ,
  expoCnt bigint ,
  clkCnt bigint
) WITH (
'connector.type' = 'elasticsearch',
'connector.version' = '6',
'connector.hosts' = '<a href="http://168.61.113.171:9092;http://168.61.113.171:9093" class="">http://168.61.113.171:9092;http://168.61.113.171:9093',
'connector.index' = 'flink_zhangle_pageview',
'connector.document-type' = '_doc',
'update-mode' = 'upsert',
'connector.key-delimiter' = '$',
'connector.key-null-literal' = 'n/a',
'connector.bulk-flush.interval' = '1000',
'format.type' = 'json'
)


And the SQL logic is as the following:

INSERT INTO ES6_ZHANGLE_OUTPUT
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clkCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min

I simply run StreamTableEnvironment.sqlUpdate( the above sql content) and execute the task. Not sure what the root cause is. 


At 2020-02-14 23:19:14, "Jark Wu" <[hidden email]> wrote:

Hi sunfulin,

Is this the real query you submit?  AFAIK, insert with column list is not allowed for now, 
i.e. the `INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)`.

Could you attach the full SQL text, including  DDLs of ES6_ZHANGLE_OUTPUT table and kafka_zl_etrack_event_stream table.
If you have a minimal program that can reproduce this problem, that would be great. 

Best,
Jark

On Fri, 14 Feb 2020 at 22:53, Robert Metzger <[hidden email]> wrote:


---------- Forwarded message ---------
From: sunfulin <[hidden email]>
Date: Fri, Feb 14, 2020 at 2:59 AM
Subject: Re:Flink 1.10 es sink exception
To: [hidden email] <[hidden email]>


Anyone can share a little advice on the reason of this exception? I changed to use old planner, the same sql runs well. 





At 2020-02-13 16:07:18, "sunfulin" <[hidden email]> wrote:

Hi, guys
When running the same Flink sql like the following, I met exception like "org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated". I am using the latest Flink 1.10 release with blink planner enabled. Because the same logic runs well within Flink 1.8.2 old planner. Does the SQL usage has some problem or may has a bug here ? 


INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clickCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min



 


 


 



 
Leonard Xu
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Flink 1.10 es sink exception

Leonard Xu
Hi, sunfulin

Using constant key in `group by` query is not usual and inefficient, you  can get around this bug by bubbling up your constant key in `group by` from now.  

BTW，godfrey is ready to resolve issue. 


在 2020年2月17日，10:15，sunfulin <[hidden email]> 写道：

Hi，
WOW，really thankful for the track and debug of this problem. I can see the constant key issue. Appreciate for your kindly help : )





At 2020-02-15 21:06:58, "Leonard Xu" <[hidden email]> wrote:


Hi, sunfulin
I reproduce your case，this should be a bug in extracting unique key from plan and I create an issue[1] to trace this.


CC:  jark


[1]https://issues.apache.org/jira/browse/FLINK-16070



在 2020年2月14日，23:39，sunfulin <[hidden email]> 写道：

Hi, Jark
Appreciate for your reply. insert with column list indeed is not allowed with old planner enabled in Flink 1.10 while it will throws exception such as "Partial insert is not supported". 
Never mind for this issue. Focus on the UpsertMode exception, my es DDL is like the following: 

CREATE TABLE ES6_ZHANGLE_OUTPUT (
  aggId varchar ,
  pageId varchar ,
  ts varchar ,
  expoCnt bigint ,
  clkCnt bigint
) WITH (
'connector.type' = 'elasticsearch',
'connector.version' = '6',
'connector.hosts' = '<a href="http://168.61.113.171:9092;http://168.61.113.171:9093" class="">http://168.61.113.171:9092;http://168.61.113.171:9093',
'connector.index' = 'flink_zhangle_pageview',
'connector.document-type' = '_doc',
'update-mode' = 'upsert',
'connector.key-delimiter' = '$',
'connector.key-null-literal' = 'n/a',
'connector.bulk-flush.interval' = '1000',
'format.type' = 'json'
)


And the SQL logic is as the following:

INSERT INTO ES6_ZHANGLE_OUTPUT
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clkCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min

I simply run StreamTableEnvironment.sqlUpdate( the above sql content) and execute the task. Not sure what the root cause is. 


At 2020-02-14 23:19:14, "Jark Wu" <[hidden email]> wrote:

Hi sunfulin,

Is this the real query you submit?  AFAIK, insert with column list is not allowed for now, 
i.e. the `INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)`.

Could you attach the full SQL text, including  DDLs of ES6_ZHANGLE_OUTPUT table and kafka_zl_etrack_event_stream table.
If you have a minimal program that can reproduce this problem, that would be great. 

Best,
Jark

On Fri, 14 Feb 2020 at 22:53, Robert Metzger <[hidden email]> wrote:


---------- Forwarded message ---------
From: sunfulin <[hidden email]>
Date: Fri, Feb 14, 2020 at 2:59 AM
Subject: Re:Flink 1.10 es sink exception
To: [hidden email] <[hidden email]>


Anyone can share a little advice on the reason of this exception? I changed to use old planner, the same sql runs well. 





At 2020-02-13 16:07:18, "sunfulin" <[hidden email]> wrote:

Hi, guys
When running the same Flink sql like the following, I met exception like "org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated". I am using the latest Flink 1.10 release with blink planner enabled. Because the same logic runs well within Flink 1.8.2 old planner. Does the SQL usage has some problem or may has a bug here ? 


INSERT INTO ES6_ZHANGLE_OUTPUT(aggId, pageId, ts, expoCnt, clkCnt)
  SELECT aggId, pageId, ts_min as ts,
  count(case when eventId = 'exposure' then 1 else null end) as expoCnt,
  count(case when eventId = 'click' then 1 else null end) as clickCnt
  FROM
  (
    SELECT
        'ZL_001' as aggId,
        pageId,
        eventId,
        recvTime,
        ts2Date(recvTime) as ts_min
    from kafka_zl_etrack_event_stream
    where eventId in ('exposure', 'click')
  ) as t1
  group by aggId, pageId, ts_min



 


 


 



 
Free forum by Nabble Edit this page