问题一：pyflink 如何使用session window对相同pv数据聚合？
hi,all： 一账号一段时间内连续操作为一个pv，间隔时间超过阈值后会记为新的pv。系统需要获取流式日志，使用日志统计实时数据的各项指标。但是我们在使用session window的时候无法使用udaf(自定义聚合函数)对相同pv日志进行聚合统计。 希望知道的大佬能给点建议。感谢！！！
session_window = Session.with_gap("60.second").on("pv_time").alias("w") t_env.from_path('source')
.window(session_window)
.group_by("w,pv_id")
.select("pv_id,get_act(act)").insert_into("sink")
http://apache-flink.147419.n8.nabble.com/file/t1355/infoflow_2021-03-08_19-02-16.png*来自志愿者整理的flink邮件归档
参考回答：
1.12 还不支持session window的udaf，在1.13上将提供这部分的支持，具体可以关注JIRA[1]。
然后，1.12是支持ProcessFunction和KeyedProcessFunction的，具体可以参考代码[2]
[1] https://issues.apache.org/jira/browse/FLINK-21630
[2]
https://github.com/apache/flink/blob/release-1.12/flink-python/pyflink/datastream/functions.py
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359520?spm=a2c6h.13262185.0.0.133c39c0Clkic3
问题二：flink作业报 task manager连接错误怎么办？
flink作业在执行几个小时候总是报某一个节点失去连接的错误，有没有朋友遇到过？ 报错日志如下: org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: Connection unexpectedly closed by remote task manager 'cdh-slave-01/130.0.0.1:41800'. This might indicate that the remote task manager was lost.*来自志愿者整理的flink邮件归档
参考回答：
你好， 可以看下具体那个 TaskManager 的日志，我之前遇到的这种情况一般都是内存用超被容器（比如Yarn）Kill 掉或者是 TaskManager 里面抛异常了。如果是 received signal 15 一般就是被容器 kill 掉了，可以看下容器的日志，其他情况可以看下具体的异常。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359521?spm=a2c6h.13262185.0.0.133c39c0Clkic3
问题三：flink dataStream多次sink DAG重复驱动执行？
有个疑问， 如下程序片段：
Properties properties = new Properties(); properties.setProperty("bootstrap.servers",kafkaAddr); properties.setProperty("group.id",kafkaOdsGroup); properties.setProperty("auto.offset.reset",kafkaOdsAutoOffsetReset);
properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS,kafkaOdsPartitionDiscoverInterval);
properties.setProperty("transaction.timeout.ms",KafkaOdsTransactionTimeout);//kafka事务超时时间
FlinkKafkaConsumer flinkKafkaConsumer = new FlinkKafkaConsumer<>(kafkaOdsTopic,new SimpleStringSchema(),properties); DataStreamSource dataStreamSource = env.addSource(flinkKafkaConsumer); dataStreamSource.printToErr("1"); dataStreamSource.printToErr("2"); dataStreamSource.printToErr("3");
我对一个datastream进行多次相同操作的sink，请问是否会导致上游整个DAG重复驱动执行，基于spark的惯性思维，我认为上游DAG是会重复驱动执行的？*来自志愿者整理的flink邮件归档
参考回答：
个人理解是不会重复驱动执行的，具体你可以测试一下，但是从底层原理上讲，我也讲不了。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359534?spm=a2c6h.13262185.0.0.133c39c0Clkic3
