Apache Flink 中文用户邮件列表

使用Kafka Sink时报错：AppendStreamTableSink doesn't support consuming update changes

classic Classic list List threaded Threaded
♦
♦
 4 messages Options
xiao cai
Reply | Threaded
Open this post in threaded view
♦
♦
|

使用Kafka Sink时报错：AppendStreamTableSink doesn't support consuming update changes

xiao cai
Hi Jark:
版本：1.11.0
问题：flink-sql，数据经过group by 和left join后写入kafka sink，会在语法校验阶段报错：
AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate


我希望能够在sql校验时也能使upsert操作作用于kafka sink，或者等upsert完成，再写入kafka


附上执行sql：
create table kafka_table_1 (  
    `shop_id` varchar,  
    `user_id` bigint,  
    `category_id` int,
    `ts` bigint,  
    `row_time` timestamp(3),
    `proc_time` timestamp(3),
) with (  
    'connector.type' = 'kafka',  
    'connector.version' = 'universal',  
    'connector.topic' = 'user_visit_1',  
    'connector.startup-mode' = 'latest-offset',  
    'connector.properties.bootstrap.servers' = 'ip:9092',  
    'connector.properties.zookeeper.connect' = 'ip:2181',  
    'update-mode' = 'append',
    'format.type' = 'avro-registry',
    'format.schema-subject' = 'user_visit',
    'format.schema-url'='<a href="http://ip:8081'">http://ip:8081',
)


CREATE TABLE hbase_table (
    rowKey STRING,
    cf ROW<age STRING, area STRING>
) WITH (
    'connector.type' = 'hbase',
    'connector.version' = '1.4.3',
    'connector.table-name' = 'hbase_table',
    'connector.zookeeper.quorum' = 'ip:2181',
    'connector.zookeeper.znode.parent' = '/hbase',
    'connector.write.buffer-flush.max-rows' = '1000'
)




create table kafka_table_2 (  
    `shop_id` varchar,  
    `age` varchar,  
    `area` varchar
) with (  
    'connector.type' = 'kafka',  
    'connector.version' = 'universal',  
    'connector.topic' = 'user_visit_2',  
    'connector.startup-mode' = 'latest-offset',  
    'connector.properties.bootstrap.servers' = 'ip:9092',  
    'connector.properties.zookeeper.connect' = 'ip:2181',  
    'update-mode' = 'append',
    'format.type' = 'avro-registry',
    'format.schema-subject' = 'user_visit',
    'format.schema-url'='<a href="http://ip:8081'">http://ip:8081',
)


insert into kafka_table_2(shop_id, user_id, category_id, ts, row_time, proc_time)
select shop_id, age, area
from kafka_table_1 left join hbase_table
for system_time as of kafka_table_1.proc_time as temp on kafka_table_1.shop_id = temp.rowKey
group by shop_id, age, area


 原始邮件
发件人: xiao cai<[hidden email]>
收件人: user-zh<[hidden email]>
发送时间: 2020年8月12日(周三) 15:41
主题: AppendStreamTableSink doesn't support consuming update changes


Hi Jark： 版本：1.11.0 问题：flink-sql，数据经过group by 和left join后写入kafka sink，会在语法校验阶段报错： AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate
徐振华
Reply | Threaded
Open this post in threaded view
♦
♦
|

回复：使用Kafka Sink时报错：AppendStreamTableSink doesn't support consuming update changes

徐振华
是不是update-mode 改用 update模式&nbsp;




------------------&nbsp;原始邮件&nbsp;------------------
发件人:                                                                                                                        "user-zh"                                                                                    <[hidden email]&gt;;
发送时间:&nbsp;2020年8月12日(星期三) 下午3:58
收件人:&nbsp;"user-zh"<[hidden email]&gt;;

主题:&nbsp;使用Kafka Sink时报错：AppendStreamTableSink doesn't support consuming update changes



Hi Jark:
版本：1.11.0
问题：flink-sql，数据经过group by 和left join后写入kafka sink，会在语法校验阶段报错：
AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate


我希望能够在sql校验时也能使upsert操作作用于kafka sink，或者等upsert完成，再写入kafka


附上执行sql：
create table kafka_table_1 (&nbsp;
&nbsp;&nbsp;&nbsp; `shop_id` varchar,&nbsp;
&nbsp;&nbsp;&nbsp; `user_id` bigint,&nbsp;
&nbsp;&nbsp;&nbsp; `category_id` int,
&nbsp;&nbsp;&nbsp; `ts` bigint,&nbsp;
&nbsp;&nbsp;&nbsp; `row_time` timestamp(3),
&nbsp;&nbsp;&nbsp; `proc_time` timestamp(3),
) with (&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.type' = 'kafka',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.version' = 'universal',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.topic' = 'user_visit_1',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.startup-mode' = 'latest-offset',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.properties.bootstrap.servers' = 'ip:9092',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.properties.zookeeper.connect' = 'ip:2181',&nbsp;
&nbsp;&nbsp;&nbsp; 'update-mode' = 'append',
&nbsp;&nbsp;&nbsp; 'format.type' = 'avro-registry',
&nbsp;&nbsp;&nbsp; 'format.schema-subject' = 'user_visit',
&nbsp;&nbsp;&nbsp; 'format.schema-url'='<a href="http://ip:8081'">http://ip:8081',
)


CREATE TABLE hbase_table (
&nbsp;&nbsp;&nbsp; rowKey STRING,
&nbsp;&nbsp;&nbsp; cf ROW<age STRING, area STRING&gt;
) WITH (
&nbsp;&nbsp;&nbsp; 'connector.type' = 'hbase',
&nbsp;&nbsp;&nbsp; 'connector.version' = '1.4.3',
&nbsp;&nbsp;&nbsp; 'connector.table-name' = 'hbase_table',
&nbsp;&nbsp;&nbsp; 'connector.zookeeper.quorum' = 'ip:2181',
&nbsp;&nbsp;&nbsp; 'connector.zookeeper.znode.parent' = '/hbase',
&nbsp;&nbsp;&nbsp; 'connector.write.buffer-flush.max-rows' = '1000'
)




create table kafka_table_2 (&nbsp;
&nbsp;&nbsp;&nbsp; `shop_id` varchar,&nbsp;
&nbsp;&nbsp;&nbsp; `age` varchar,&nbsp;
&nbsp;&nbsp;&nbsp; `area` varchar
) with (&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.type' = 'kafka',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.version' = 'universal',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.topic' = 'user_visit_2',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.startup-mode' = 'latest-offset',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.properties.bootstrap.servers' = 'ip:9092',&nbsp;
&nbsp;&nbsp;&nbsp; 'connector.properties.zookeeper.connect' = 'ip:2181',&nbsp;
&nbsp;&nbsp;&nbsp; 'update-mode' = 'append',
&nbsp;&nbsp;&nbsp; 'format.type' = 'avro-registry',
&nbsp;&nbsp;&nbsp; 'format.schema-subject' = 'user_visit',
&nbsp;&nbsp;&nbsp; 'format.schema-url'='<a href="http://ip:8081'">http://ip:8081',
)


insert into kafka_table_2(shop_id, user_id, category_id, ts, row_time, proc_time)
select shop_id, age, area
from kafka_table_1 left join hbase_table
for system_time as of kafka_table_1.proc_time as temp on kafka_table_1.shop_id = temp.rowKey
group by shop_id, age, area


&nbsp;原始邮件
发件人: xiao cai<[hidden email]&gt;
收件人: user-zh<[hidden email]&gt;
发送时间: 2020年8月12日(周三) 15:41
主题: AppendStreamTableSink doesn't support consuming update changes


Hi Jark： 版本：1.11.0 问题：flink-sql，数据经过group by 和left join后写入kafka sink，会在语法校验阶段报错： AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate
Leonard Xu
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: 使用Kafka Sink时报错：AppendStreamTableSink doesn't support consuming update changes

Leonard Xu
In reply to this post by xiao cai
Hi

Group by 和 left join 都是会有 retract 消息的，这类消息需要UpsertStreamTableSink才能处理， Kafka connetor 目前的实现是AppendStreamTableSink，所以不能处理
社区已经有一个issue在处理这个问题了，应该1.12能提供这个功能。


Best
Leonard
[1]https://issues.apache.org/jira/browse/FLINK-18826 <https://issues.apache.org/jira/browse/FLINK-18826>

> 在 2020年8月12日，15:58，xiao cai <[hidden email]> 写道：
>
> Hi Jark:
> 版本：1.11.0
> 问题：flink-sql，数据经过group by 和left join后写入kafka sink，会在语法校验阶段报错：
> AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate
>
>
> 我希望能够在sql校验时也能使upsert操作作用于kafka sink，或者等upsert完成，再写入kafka
>
>
> 附上执行sql：
> create table kafka_table_1 (  
>    `shop_id` varchar,  
>    `user_id` bigint,  
>    `category_id` int,
>    `ts` bigint,  
>    `row_time` timestamp(3),
>    `proc_time` timestamp(3),
> ) with (  
>    'connector.type' = 'kafka',  
>    'connector.version' = 'universal',  
>    'connector.topic' = 'user_visit_1',  
>    'connector.startup-mode' = 'latest-offset',  
>    'connector.properties.bootstrap.servers' = 'ip:9092',  
>    'connector.properties.zookeeper.connect' = 'ip:2181',  
>    'update-mode' = 'append',
>    'format.type' = 'avro-registry',
>    'format.schema-subject' = 'user_visit',
>    'format.schema-url'='<a href="http://ip:8081'">http://ip:8081',
> )
>
>
> CREATE TABLE hbase_table (
>    rowKey STRING,
>    cf ROW<age STRING, area STRING>
> ) WITH (
>    'connector.type' = 'hbase',
>    'connector.version' = '1.4.3',
>    'connector.table-name' = 'hbase_table',
>    'connector.zookeeper.quorum' = 'ip:2181',
>    'connector.zookeeper.znode.parent' = '/hbase',
>    'connector.write.buffer-flush.max-rows' = '1000'
> )
>
>
>
>
> create table kafka_table_2 (  
>    `shop_id` varchar,  
>    `age` varchar,  
>    `area` varchar
> ) with (  
>    'connector.type' = 'kafka',  
>    'connector.version' = 'universal',  
>    'connector.topic' = 'user_visit_2',  
>    'connector.startup-mode' = 'latest-offset',  
>    'connector.properties.bootstrap.servers' = 'ip:9092',  
>    'connector.properties.zookeeper.connect' = 'ip:2181',  
>    'update-mode' = 'append',
>    'format.type' = 'avro-registry',
>    'format.schema-subject' = 'user_visit',
>    'format.schema-url'='<a href="http://ip:8081'">http://ip:8081',
> )
>
>
> insert into kafka_table_2(shop_id, user_id, category_id, ts, row_time, proc_time)
> select shop_id, age, area
> from kafka_table_1 left join hbase_table
> for system_time as of kafka_table_1.proc_time as temp on kafka_table_1.shop_id = temp.rowKey
> group by shop_id, age, area
>
>
> 原始邮件
> 发件人: xiao cai<[hidden email]>
> 收件人: user-zh<[hidden email]>
> 发送时间: 2020年8月12日(周三) 15:41
> 主题: AppendStreamTableSink doesn't support consuming update changes
>
>
> Hi Jark： 版本：1.11.0 问题：flink-sql，数据经过group by 和left join后写入kafka sink，会在语法校验阶段报错： AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate

xiao cai
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: 使用Kafka Sink时报错：AppendStreamTableSink doesn't support consuming update changes

xiao cai
In reply to this post by xiao cai
Dear Leonard Xu:
我会去关注这个issue，非常感谢答疑。


 原始邮件
发件人: Leonard Xu<[hidden email]>
收件人: user-zh<[hidden email]>
发送时间: 2020年8月12日(周三) 16:05
主题: Re: 使用Kafka Sink时报错：AppendStreamTableSink doesn't support consuming update changes


Hi Group by 和 left join 都是会有 retract 消息的，这类消息需要UpsertStreamTableSink才能处理， Kafka connetor 目前的实现是AppendStreamTableSink，所以不能处理 社区已经有一个issue在处理这个问题了，应该1.12能提供这个功能。 Best Leonard [1]https://issues.apache.org/jira/browse/FLINK-18826 <">https://issues.apache.org/jira/browse/FLINK-18826> > 在 2020年8月12日，15:58，xiao cai <[hidden email]> 写道： > > Hi Jark: > 版本：1.11.0 > 问题：flink-sql，数据经过group by 和left join后写入kafka sink，会在语法校验阶段报错： > AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate > > > 我希望能够在sql校验时也能使upsert操作作用于kafka sink，或者等upsert完成，再写入kafka > > > 附上执行sql： > create table kafka_table_1 ( > `shop_id` varchar, > `user_id` bigint, > `category_id` int, > `ts` bigint, > `row_time` timestamp(3), > `proc_time` timestamp(3), > ) with ( > 'connector.type' = 'kafka', > 'connector.version' = 'universal', > 'connector.topic' = 'user_visit_1', > 'connector.startup-mode' = 'latest-offset', > 'connector.properties.bootstrap.servers' = 'ip:9092', > 'connector.properties.zookeeper.connect' = 'ip:2181', > 'update-mode' = 'append', > 'format.type' = 'avro-registry', > 'format.schema-subject' = 'user_visit', > 'format.schema-url'='<a href="http://ip:8081'">http://ip:8081', > ) > > > CREATE TABLE hbase_table ( > rowKey STRING, > cf ROW<age STRING, area STRING> > ) WITH ( > 'connector.type' = 'hbase', > 'connector.version' = '1.4.3', > 'connector.table-name' = 'hbase_table', > 'connector.zookeeper.quorum' = 'ip:2181', > 'connector.zookeeper.znode.parent' = '/hbase', > 'connector.write.buffer-flush.max-rows' = '1000' > ) > > > > > create table kafka_table_2 ( > `shop_id` varchar, > `age` varchar, > `area` varchar > ) with ( > 'connector.type' = 'kafka', > 'connector.version' = 'universal', > 'connector.topic' = 'user_visit_2', > 'connector.startup-mode' = 'latest-offset', > 'connector.properties.bootstrap.servers' = 'ip:9092', > 'connector.properties.zookeeper.connect' = 'ip:2181', > 'update-mode' = 'append', > 'format.type' = 'avro-registry', > 'format.schema-subject' = 'user_visit', > 'format.schema-url'='<a href="http://ip:8081'">http://ip:8081', > ) > > > insert into kafka_table_2(shop_id, user_id, category_id, ts, row_time, proc_time) > select shop_id, age, area > from kafka_table_1 left join hbase_table > for system_time as of kafka_table_1.proc_time as temp on kafka_table_1.shop_id = temp.rowKey > group by shop_id, age, area > > > 原始邮件 > 发件人: xiao cai<[hidden email]> > 收件人: user-zh<[hidden email]> > 发送时间: 2020年8月12日(周三) 15:41 > 主题: AppendStreamTableSink doesn't support consuming update changes > > > Hi Jark： 版本：1.11.0 问题：flink-sql，数据经过group by 和left join后写入kafka sink，会在语法校验阶段报错： AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate
Free forum by Nabble Edit this page