You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by kcz <57...@qq.com.INVALID> on 2021/09/22 03:41:20 UTC

flink-1.12.0 流模式 使用 lag问题

如何使用才是正确的，求大佬帮看看
behavior,next_bv 字段内容一直是保持一致的，无法得到自己想要的结果


发送的数据
{
&nbsp;&nbsp;"user_id":&nbsp;1,
&nbsp;&nbsp;"item_id":&nbsp;1,
&nbsp;&nbsp;"behavior":"pv1"
}
{
&nbsp;&nbsp;"user_id":&nbsp;1,
&nbsp;&nbsp;"item_id":&nbsp;1,
&nbsp;&nbsp;"behavior":"pv2"
}






CREATE TABLE KafkaTable (
&nbsp; `user_id` BIGINT,
&nbsp; `item_id` BIGINT,
&nbsp; `behavior` STRING,
&nbsp; proctime as PROCTIME()
) WITH (
&nbsp; 'connector' = 'kafka',
&nbsp; 'topic' = 'user_behavior',
&nbsp; 'properties.bootstrap.servers' = '',
&nbsp; 'properties.group.id' = 'testGroup',
&nbsp; 'scan.startup.mode' = 'earliest-offset',
&nbsp; 'format' = 'json'
);



SELECT
	user_id,
	item_id,
	behavior,
	next_bv&nbsp;
FROM
	( SELECT *, lag( behavior, 1 ) over ( PARTITION BY user_id ORDER BY proctime ) AS next_bv FROM KafkaTable ) t;

回复： flink-1.12.0 流模式 使用 lag问题

Posted by kcz <57...@qq.com.INVALID>.
tks，我去换版本使用下




------------------&nbsp;原始邮件&nbsp;------------------
发件人:                                                                                                                        "user-zh"                                                                                    <libenchao@apache.org&gt;;
发送时间:&nbsp;2021年9月22日(星期三) 中午12:04
收件人:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;;

主题:&nbsp;Re: flink-1.12.0 流模式 使用 lag问题



Hi，这个是一个已知的bug[1]，已经在1.13.1以及1.4版本修复了。
可以使用一下1.13.1试一下，1.4版本现在也正在投票中了，应该很快就会发布出来了。

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

kcz <573693104@qq.com.invalid&gt; 于2021年9月22日周三 上午11:41写道：

&gt; 如何使用才是正确的，求大佬帮看看
&gt; behavior,next_bv 字段内容一直是保持一致的，无法得到自己想要的结果
&gt;
&gt;
&gt; 发送的数据
&gt; {
&gt; &amp;nbsp;&amp;nbsp;"user_id":&amp;nbsp;1,
&gt; &amp;nbsp;&amp;nbsp;"item_id":&amp;nbsp;1,
&gt; &amp;nbsp;&amp;nbsp;"behavior":"pv1"
&gt; }
&gt; {
&gt; &amp;nbsp;&amp;nbsp;"user_id":&amp;nbsp;1,
&gt; &amp;nbsp;&amp;nbsp;"item_id":&amp;nbsp;1,
&gt; &amp;nbsp;&amp;nbsp;"behavior":"pv2"
&gt; }
&gt;
&gt;
&gt;
&gt;
&gt;
&gt;
&gt; CREATE TABLE KafkaTable (
&gt; &amp;nbsp; `user_id` BIGINT,
&gt; &amp;nbsp; `item_id` BIGINT,
&gt; &amp;nbsp; `behavior` STRING,
&gt; &amp;nbsp; proctime as PROCTIME()
&gt; ) WITH (
&gt; &amp;nbsp; 'connector' = 'kafka',
&gt; &amp;nbsp; 'topic' = 'user_behavior',
&gt; &amp;nbsp; 'properties.bootstrap.servers' = '',
&gt; &amp;nbsp; 'properties.group.id' = 'testGroup',
&gt; &amp;nbsp; 'scan.startup.mode' = 'earliest-offset',
&gt; &amp;nbsp; 'format' = 'json'
&gt; );
&gt;
&gt;
&gt;
&gt; SELECT
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; user_id,
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; item_id,
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; behavior,
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; next_bv&amp;nbsp;
&gt; FROM
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ( SELECT *, lag( behavior, 1 ) over ( PARTITION BY user_id ORDER
&gt; BY proctime ) AS next_bv FROM KafkaTable ) t;



-- 

Best,
Benchao Li

Re: flink-1.12.0 流模式 使用 lag问题

Posted by Benchao Li <li...@apache.org>.
Hi，这个是一个已知的bug[1]，已经在1.13.1以及1.4版本修复了。
可以使用一下1.13.1试一下，1.4版本现在也正在投票中了，应该很快就会发布出来了。

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

kcz <57...@qq.com.invalid> 于2021年9月22日周三 上午11:41写道：

> 如何使用才是正确的，求大佬帮看看
> behavior,next_bv 字段内容一直是保持一致的，无法得到自己想要的结果
>
>
> 发送的数据
> {
> &nbsp;&nbsp;"user_id":&nbsp;1,
> &nbsp;&nbsp;"item_id":&nbsp;1,
> &nbsp;&nbsp;"behavior":"pv1"
> }
> {
> &nbsp;&nbsp;"user_id":&nbsp;1,
> &nbsp;&nbsp;"item_id":&nbsp;1,
> &nbsp;&nbsp;"behavior":"pv2"
> }
>
>
>
>
>
>
> CREATE TABLE KafkaTable (
> &nbsp; `user_id` BIGINT,
> &nbsp; `item_id` BIGINT,
> &nbsp; `behavior` STRING,
> &nbsp; proctime as PROCTIME()
> ) WITH (
> &nbsp; 'connector' = 'kafka',
> &nbsp; 'topic' = 'user_behavior',
> &nbsp; 'properties.bootstrap.servers' = '',
> &nbsp; 'properties.group.id' = 'testGroup',
> &nbsp; 'scan.startup.mode' = 'earliest-offset',
> &nbsp; 'format' = 'json'
> );
>
>
>
> SELECT
>         user_id,
>         item_id,
>         behavior,
>         next_bv&nbsp;
> FROM
>         ( SELECT *, lag( behavior, 1 ) over ( PARTITION BY user_id ORDER
> BY proctime ) AS next_bv FROM KafkaTable ) t;



-- 

Best,
Benchao Li