Re:回复：flink对象重用问题询问

Thu, 29 Aug 2024 00:18:36 -0700 

Hi,

应该是可以的。




--

    Best！
    Xuyang





在 2024-08-29 15:00:54，"刘仲诺" <[email protected]> 写道：
>您好，我目前使用的是DataStream API，代码如下：
>public class BytesProcessor extends ProcessWindowFunction<Tuple2<Long, 
>Long&gt;, BytesResult, String, TimeWindow&gt; {
>&nbsp; &nbsp; private final BytesResult bytesResult = new BytesResult();
>&nbsp; &nbsp; @Override
>&nbsp; &nbsp; public void process(String key,
>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
>&nbsp; ProcessWindowFunction<Tuple2<Long, Long&gt;, BytesResult, String, 
>TimeWindow&gt;.Context context,
>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
>&nbsp; Iterable<Tuple2<Long, Long&gt;&gt; accumulator,
>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
>&nbsp; Collector<BytesResult&gt; out) throws Exception {
>&nbsp; &nbsp; &nbsp; &nbsp; Tuple2<Long, Long&gt; result = 
>accumulator.iterator().next();
>//&nbsp; &nbsp; &nbsp; &nbsp; BytesResult bytesResult = new BytesResult();
>&nbsp; &nbsp; &nbsp; &nbsp; bytesResult.setChannelId(key);
>&nbsp; &nbsp; &nbsp; &nbsp; bytesResult.setLastHopRecvBytes(result.f0);
>&nbsp; &nbsp; &nbsp; &nbsp; bytesResult.setNextHopSendBytes(result.f1);
>&nbsp; &nbsp; &nbsp; &nbsp; 
>bytesResult.setCurrentTime(context.window().getStart());
>&nbsp; &nbsp; &nbsp; &nbsp; if (!Objects.equals(result.f0, result.f1)) {
>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; out.collect(bytesResult);
>&nbsp; &nbsp; &nbsp; &nbsp; }
>&nbsp; &nbsp; }
>}
>
>我在这里用算子实例的bytesResult变量代替了每次输出新建的bytesResult，这样的做法是否安全呢
>
>
>
>
>------------------&nbsp;原始邮件&nbsp;------------------
>发件人:                                                                           
>                                             "user-zh"                         
>                                                           
><[email protected]&gt;;
>发送时间:&nbsp;2024年8月29日(星期四) 下午2:56
>收件人:&nbsp;"user-zh"<[email protected]&gt;;
>
>主题:&nbsp;Re:flink对象重用问题询问
>
>
>
>Hi,
>
>在算子output时，是可以复用row的，在flink sql所转化的operator中，就有很多类似的做法，具体可以参考[1][2]
>
>
>
>
>[1] 
>https://github.com/apache/flink/blob/576ec2b9361a3f8d58fb22b998b0ca7c3c8cf10e/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperator.java#L45
>
>[2] 
>https://github.com/apache/flink/blob/576ec2b9361a3f8d58fb22b998b0ca7c3c8cf10e/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/GroupAggFunction.java#L66
>
>
>
>
>--
>
>&nbsp;&nbsp;&nbsp; Best！
>&nbsp;&nbsp;&nbsp; Xuyang
>
>
>
>
>
>在 2024-08-29 12:42:28，"刘仲诺" <[email protected]&gt; 写道：
>&gt;您好，目前我正在开发flink流计算程序，请问flink在算子函数中是否允许对象重用呢，比如我在构造输出记录时，不是每次都新建一条记录，而是只新建一个记录对象作为算子函数实例的成员，在构造输出记录时只更改这个对象的属性然后进行输出，请问这样的做法在flink中是安全的吗？我想进行这种操作的主要目的是减少对象创建

回复