Re:Re: Re: [Internet]Re: Re: Some question with Flink state
Hi, 理论上来说这句话是不是有问题? > “是因为如果使用value-state,一个task会存在多个key,不同的key的内容会进行替换” 因为ValueState也是keyedState的一种,所以也是每个key各自维护一个valuestate,不同的key之间是隔离的。 其实一般情况下ValueState里面存Map,和直接MapState没啥区别,只不过在不同的状态存储上和状态的TTL策略有略微不同,所以不太推荐ValueState里面存Map。 所以其实还是看具体的业务场景,假如只是算一个累加的值的话,用valuestate就够了。 -- Best! Xuyang 在 2022-05-25 13:38:52,"lxk7...@163.com" 写道: > >刚看了下keygroup的原理,前面的内容大致能理解了,对于下面这段话 >"map-state的话相当于某些固定的key group里面的key都可以通过map-state的user-key去分别存储" >我理解 >是因为如果使用value-state,一个task会存在多个key,不同的key的内容会进行替换,而使用map的话,就算同一个task有多个key,根据用户自定义的key还是可以匹配到的。 >这样的话,大部分场景其实都适合使用map-state。 > > >lxk7...@163.com > >From: jurluo(罗凯) >Date: 2022-05-25 11:05 >To: user-zh@flink.apache.org >Subject: Re: [Internet]Re: Re: Some question with Flink state >老哥,看起来好像没什么问题,相同的key都分配在了同个task,每个task会存在多种key是正常的。key会按最大并行度分成多个key >group,然后固定的key >group分配到各个task里。只能保证相同的key会到同一个task,不能保证一个task只有一个key。你这个需求用map-state才合适。map-state的话相当于某些固定的key > group里面的key都可以通过map-state的user-key去分别存储。 > >> 2022年5月25日 上午10:45,lxk7...@163.com 写道: >> >> 图片好像又挂了 我重发下 >> hello,我这边测试了一下发现一个问题,在使用String类型做keyby的时候并没有得到正确的结果,而使用int类型的时候结果是正确。而且测试发现两次keyby确实是以第二次keyby为准 >> >> >> >>下面是我的代码及测试结果 >> >> >> >> 一.使用int类型 >> >> >> >>public class KeyByTest { >> >> >> >> public static void main(String[] args) throws Exception { >> >> >> >> StreamExecutionEnvironment env = >> StreamExecutionEnvironment.getExecutionEnvironment(); >> >> >> >> env.setParallelism(10); >> >> >> >> >> >> DataStreamSource dataDataStreamSource = >> env.fromCollection(Arrays.asList(new data(1, "123", "首页"), >> >> >> >> new data(1, "123", "分类页"), >> >> >> >> new data(2, "r-123", "搜索结果页"), >> >> >> >> new data(1, "r-123", "我的页"), >> >> >> >> new data(3, "r-4567", "搜索结果页"))); >> >> >> >> >> >> >> >> >> >> >> >> SingleOutputStreamOperator map = >> dataDataStreamSource.keyBy(new MyKeySelector()) >> >> >> >> .map(new RichMapFunction() { >> >> >> >> >> >> @Override >> >> >> >> public String map(data data) throws Exception { >> >> >> >> System.out.println(data.toString() + "的subtask为:" + >> getRuntimeContext().getIndexOfThisSubtask() ); >> >> >> >> return data.toString(); >> >> >> >> } >> >> >> >> }); >> >> >> >> >> >> >> >> env.execute("test"); >> >> >> >> >> >> } >> >> >> >> } >> >> >> >> class data{ >> >> >> >> private int id; >> >> >> >> private String goods; >> >> >> >> private String pageName; >> >> >> >> >> >> public data(int id, String goods, String pageName) { >> >> >> >> this.id = id; >> >> >> >> this.goods = goods; >> >> >> >> this.pageName = pageName; >> >> >> >> } >> >> >> >> >> >> >> >> public data() { >> >> >> >> } >> >> >> >> >> >> public int getId() { >> >> >> >> return id; >> >> >> >> } >> >> >> >> >> >> public void setId(int id) { >> >> >> >> this.id = id; >> >> >> >> } >> >> >> >> >> >> public String getGoods() { >> >> >> >> return goods; >> >> >> >> } >> >> >> >> >> >> public void setGoods(String goods) { >> >> >> >> this.goods = goods;
Re: Re: [Internet]Re: Re: Some question with Flink state
刚看了下keygroup的原理,前面的内容大致能理解了,对于下面这段话 "map-state的话相当于某些固定的key group里面的key都可以通过map-state的user-key去分别存储" 我理解 是因为如果使用value-state,一个task会存在多个key,不同的key的内容会进行替换,而使用map的话,就算同一个task有多个key,根据用户自定义的key还是可以匹配到的。 这样的话,大部分场景其实都适合使用map-state。 lxk7...@163.com From: jurluo(罗凯) Date: 2022-05-25 11:05 To: user-zh@flink.apache.org Subject: Re: [Internet]Re: Re: Some question with Flink state 老哥,看起来好像没什么问题,相同的key都分配在了同个task,每个task会存在多种key是正常的。key会按最大并行度分成多个key group,然后固定的key group分配到各个task里。只能保证相同的key会到同一个task,不能保证一个task只有一个key。你这个需求用map-state才合适。map-state的话相当于某些固定的key group里面的key都可以通过map-state的user-key去分别存储。 > 2022年5月25日 上午10:45,lxk7...@163.com 写道: > > 图片好像又挂了 我重发下 > hello,我这边测试了一下发现一个问题,在使用String类型做keyby的时候并没有得到正确的结果,而使用int类型的时候结果是正确。而且测试发现两次keyby确实是以第二次keyby为准 > > > >下面是我的代码及测试结果 > > > > 一.使用int类型 > > > >public class KeyByTest { > > > > public static void main(String[] args) throws Exception { > > > > StreamExecutionEnvironment env = > StreamExecutionEnvironment.getExecutionEnvironment(); > > > > env.setParallelism(10); > > > > > > DataStreamSource dataDataStreamSource = > env.fromCollection(Arrays.asList(new data(1, "123", "首页"), > > > > new data(1, "123", "分类页"), > > > > new data(2, "r-123", "搜索结果页"), > > > > new data(1, "r-123", "我的页"), > > > > new data(3, "r-4567", "搜索结果页"))); > > > > > > > > > > > > SingleOutputStreamOperator map = > dataDataStreamSource.keyBy(new MyKeySelector()) > > > > .map(new RichMapFunction() { > > > > > > @Override > > > > public String map(data data) throws Exception { > > > > System.out.println(data.toString() + "的subtask为:" + > getRuntimeContext().getIndexOfThisSubtask() ); > > > > return data.toString(); > > > > } > > > > }); > > > > > > > > env.execute("test"); > > > > > > } > > > > } > > > > class data{ > > > > private int id; > > > > private String goods; > > > > private String pageName; > > > > > > public data(int id, String goods, String pageName) { > > > > this.id = id; > > > > this.goods = goods; > > > > this.pageName = pageName; > > > > } > > > > > > > > public data() { > > > > } > > > > > > public int getId() { > > > > return id; > > > > } > > > > > > public void setId(int id) { > > > > this.id = id; > > > > } > > > > > > public String getGoods() { > > > > return goods; > > > > } > > > > > > public void setGoods(String goods) { > > > > this.goods = goods; > > > > } > > > > > > public String getPageName() { > > > > return pageName; > > > > } > > > > > > public void setPageName(String pageName) { > > > > this.pageName = pageName; > > > > } > > > > > > @Override > > > > public String toString() { > > > > return "data{" + > > > > "id='" + id + '\'' + > > > > ", goods='" + goods + '\'' + > > > > ", pageName='" + pageName + '\'' + > > > > '}'; > > > > } > > > > } > > > > > > class MyKeySelector implements KeySelector{ > > > > > > @Override > > > > public Integer getKey(data data) throws Exception { > > > > return data.getId(); > > > > } > > > > } > > > > 控制台的输出如下: > https://s2.loli.net/2022/05/25/mxtZu9YAPN2FD1a.png > > &g
Re: [Internet]Re: Re: Some question with Flink state
老哥,看起来好像没什么问题,相同的key都分配在了同个task,每个task会存在多种key是正常的。key会按最大并行度分成多个key group,然后固定的key group分配到各个task里。只能保证相同的key会到同一个task,不能保证一个task只有一个key。你这个需求用map-state才合适。map-state的话相当于某些固定的key group里面的key都可以通过map-state的user-key去分别存储。 > 2022年5月25日 上午10:45,lxk7...@163.com 写道: > > 图片好像又挂了 我重发下 > hello,我这边测试了一下发现一个问题,在使用String类型做keyby的时候并没有得到正确的结果,而使用int类型的时候结果是正确。而且测试发现两次keyby确实是以第二次keyby为准 > > > >下面是我的代码及测试结果 > > > > 一.使用int类型 > > > >public class KeyByTest { > > > > public static void main(String[] args) throws Exception { > > > > StreamExecutionEnvironment env = > StreamExecutionEnvironment.getExecutionEnvironment(); > > > > env.setParallelism(10); > > > > > > DataStreamSource dataDataStreamSource = > env.fromCollection(Arrays.asList(new data(1, "123", "首页"), > > > > new data(1, "123", "分类页"), > > > > new data(2, "r-123", "搜索结果页"), > > > > new data(1, "r-123", "我的页"), > > > > new data(3, "r-4567", "搜索结果页"))); > > > > > > > > > > > > SingleOutputStreamOperator map = > dataDataStreamSource.keyBy(new MyKeySelector()) > > > > .map(new RichMapFunction() { > > > > > > @Override > > > > public String map(data data) throws Exception { > > > > System.out.println(data.toString() + "的subtask为:" + > getRuntimeContext().getIndexOfThisSubtask() ); > > > > return data.toString(); > > > > } > > > > }); > > > > > > > > env.execute("test"); > > > > > > } > > > > } > > > > class data{ > > > > private int id; > > > > private String goods; > > > > private String pageName; > > > > > > public data(int id, String goods, String pageName) { > > > > this.id = id; > > > > this.goods = goods; > > > > this.pageName = pageName; > > > > } > > > > > > > > public data() { > > > > } > > > > > > public int getId() { > > > > return id; > > > > } > > > > > > public void setId(int id) { > > > > this.id = id; > > > > } > > > > > > public String getGoods() { > > > > return goods; > > > > } > > > > > > public void setGoods(String goods) { > > > > this.goods = goods; > > > > } > > > > > > public String getPageName() { > > > > return pageName; > > > > } > > > > > > public void setPageName(String pageName) { > > > > this.pageName = pageName; > > > > } > > > > > > @Override > > > > public String toString() { > > > > return "data{" + > > > > "id='" + id + '\'' + > > > > ", goods='" + goods + '\'' + > > > > ", pageName='" + pageName + '\'' + > > > > '}'; > > > > } > > > > } > > > > > > class MyKeySelector implements KeySelector{ > > > > > > @Override > > > > public Integer getKey(data data) throws Exception { > > > > return data.getId(); > > > > } > > > > } > > > > 控制台的输出如下: > https://s2.loli.net/2022/05/25/mxtZu9YAPN2FD1a.png > > > > 可以看见数据根据id分组,分到了不同的subtask上。 > > > > > > > > 二.使用String类型 代码如下: > > > > public class KeyByTest { > > > > public static void main(String[] args) throws Exception { > > > > StreamExecutionEnvironment env = > StreamExecutionEnvironment.getExecutionEnvironment(); > > > > env.setParallelism(10); > > > > > > DataStreamSource dataDataStreamSource = > env.fromCollection(Arrays.asList(new data("1", "123", "首页"), > > > > new data("1", "123", "分类页"), > > > > new data("2", "r-123", "搜索结果页"), > > > > new data("2", "r-123", "我的页"), > > > > new data("3", "r-4567", "搜索结果页"))); > > > > > > > > > > > > SingleOutputStreamOperator map = > dataDataStreamSource.keyBy(new MyKeySelector()) > > > > .map(new RichMapFunction() { > > > > > > @Override > > > > public String map(data data) throws Exception { > > > > System.out.println(data.toString() + "的subtask为:" + > getRuntimeContext().getIndexOfThisSubtask() ); > > > > return data.toString(); > > > > } > > > > }); > > > > > > > > env.execute("test"); > > > > > > } > > > > } > > > > class data{ > > > > private String id; > > > > private String goods; > > > > private String pageName; > > > > > > public data(String id, String goods, String pageName) { > > > > this.id = id