date:20190304

Re: jobmanager重启之后，job没有重启

2019-03-04 文章孙森

Hi 韩非：
我这边使用flink run 
-m的方式提交的job，如果配置为zookeeper就会提交失败，所以才改为这种方式。我的jobmanager也可以成功切换，但是之前的运行的job并没有成功重启。你那边有观察Job的重启情况吗？

Best
Sen

> 在 2019年3月4日，下午10:01，Fei Han  写道：
> 
> 杨森：
>你好！对于Flink on yarn 模式 flink-conf.yaml中的配置如下：
> high-availability: zookeeper
> 
> high-availability.zookeeper.quorum: cdh1:2181,cdh2:2181,cdh3:2181
> 
> high-availability.storageDir: hdfs://cdh1:9000/flink/recovery
> 
> high-availability.zookeeper.path.root: /flink
> 
> state.backend: filesystem
> 
> state.backend.fs.checkpointdir: hdfs://cdh1:9000/flink/checkpoints
> 
> fs.hdfs.hadoopconf: /usr/local/hadoop-2.7.4/etc/hadoop
> 
> yarn-site.xml中的配置如下：
> 
> 
> 
> 
> yarn.resourcemanager.am.max-attempts
> 
> 4
> 
> 
> 
> 以上的默认配置，你也可以在程序中配置，默认配置和程序都配置试试。
> 
> 
> 
> 如图看到Attempt ID由01变为02，进程也有所变化，，说明HA切换成功了。
> 
>  希望这些内容帮助到你。
> 
>   Best,
> 
>   韩非
> 
> 
> 
> --
> 发件人：孙森 
> 发送时间：2019年3月4日(星期一) 18:41
> 收件人：user-zh 
> 主　题：jobmanager重启之后，job没有重启
> 
> 大家好：
>   有配置过flink ha吗？我配置了yarn高可用之后，kill掉job manager 
> 之后，jobmanager可以成功重启，但是yarn 
> session上正在运行的job并没有重启。我程序是开启了checkpoint的。有谁遇到过吗？是我配置的问题还是什么原因？
> 
> 附上高可用配置：
> 
> high-availability.storageDir: hdfs:///flink/ha/
> high-availability.zookeeper.quorum: h1:2181,h2:2181,h3:2181
> high-availability.zookeeper.path.root: /flink
> yarn.application-attempts: 2
> 
> 
> Thanks!
> 
>

Re: 如何每五分钟统计一次当天某个消息的总条数

2019-03-04 文章戴嘉诚

当天的，就直接是翻滚窗口就行了吧，不过你要注意你一天量有多大，小心内存不够了

张作峰 于2019年3月5日 周二15:06写道：

> 设置event time 窗口为一天，是滑动窗口吗？具体是指？需要统计的是当天的
>
> --
> 张作峰
> 创维 一体机软件开发部
>
> 深圳市南山区高新南一道创维大厦A座12楼
> 手机: 18320872958 座机: 0755-26974350（分机号 4350）
> Email：m...@zhangzuofeng.cn
> 主页：http://www.zhangzuofeng.cn
> wiki: http://wiki.qiannuo.me
>
>
>
>
>
>
>
>
>
> -- 原始邮件 --
> 发件人: "Paul Lam";
> 发送时间: 2019年3月5日(星期二) 下午2:46
> 收件人: "user-zh";
> 主题: Re: 如何每五分钟统计一次当天某个消息的总条数
>
>
>
> Hi,
>
> 你可以试下设置 event time 窗口为一天，然后设置 processing time timer 来定时每 5 分钟触发输出当天最新的结果。
>
> Best,
> Paul Lam
>
> > 在 2019年3月5日，13:16，张作峰  写道：
> >
> > 大家好！
> > 请教下诸位大牛，如何使用stream api每五分钟统计一次当天某个消息的总条数？
> > 谢谢!

?????? ????????????????????????????????????????

2019-03-04 文章 ??????

event time ??
 
--
?? 
 

??A??12??
: 18320872958 : 0755-26974350 4350??
Email??m...@zhangzuofeng.cn
??http://www.zhangzuofeng.cn
wiki: http://wiki.qiannuo.me




 




--  --
??: "Paul Lam"; 
: 2019??3??5??(??) 2:46
??: "user-zh"; 
: Re: 



Hi,

?? event time  processing time timer  5 


Best,
Paul Lam

> ?? 2019??3??513:16  ??
> 
> 
> stream api??
> !

Re: 如何每五分钟统计一次当天某个消息的总条数

2019-03-04 文章 Paul Lam

Hi,

你可以试下设置 event time 窗口为一天，然后设置 processing time timer 来定时每 5 分钟触发输出当天最新的结果。

Best,
Paul Lam

> 在 2019年3月5日，13:16，张作峰  写道：
> 
> 大家好！
> 请教下诸位大牛，如何使用stream api每五分钟统计一次当天某个消息的总条数？
> 谢谢!

Re: 如何每五分钟统计一次当天某个消息的总条数

2019-03-04 文章刘文


处理这个问题，我有一些想法：

).Flink Stream默认是处理增量数据，对指定间隔时间或数量内的数据进行分析
).可以自定义 
ProcessAllWindowFunction,相当于，对于一个Window的数据，自己实现处理逻辑，参数是在Window之前的operator也是已经处理的
).对于你，需要存储每次计算的结果，并更新到存储中心供每次计算使用(如Redis、等)
).下面是一个ProcessAllWIndowFunction的示例，供参考(实现功能: WordCount 程序(增量按单词升序排序)  )







package 
com.opensourceteams.module.bigdata.flink.example.stream.worldcount.nc.sort

import java.time.ZoneId

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
import 
org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.fs.bucketing.{BucketingSink, 
DateTimeBucketer}
import org.apache.flink.util.Collector

import scala.collection.mutable

/**
  * nc -lk 1234  输入数据
  */
object SocketWindowWordCountLocalSinkHDFSAndWindowAllAndSorted {


  def getConfiguration(isDebug:Boolean = false):Configuration={

val configuration : Configuration = new Configuration()

if(isDebug){
  val timeout = "10 s"
  val timeoutHeartbeatPause = "100 s"
  configuration.setString("akka.ask.timeout",timeout)
  configuration.setString("akka.lookup.timeout",timeout)
  configuration.setString("akka.tcp.timeout",timeout)
  configuration.setString("akka.transport.heartbeat.interval",timeout)
  
configuration.setString("akka.transport.heartbeat.pause",timeoutHeartbeatPause)
  configuration.setString("akka.watch.heartbeat.pause",timeout)
  configuration.setInteger("heartbeat.interval",1000)
  configuration.setInteger("heartbeat.timeout",5000)
}


configuration
  }

  def main(args: Array[String]): Unit = {


val port = 1234
// get the execution environment
   // val env: StreamExecutionEnvironment = 
StreamExecutionEnvironment.getExecutionEnvironment


val configuration : Configuration = getConfiguration(true)

val env:StreamExecutionEnvironment = 
StreamExecutionEnvironment.createLocalEnvironment(1,configuration)






// get input data by connecting to the socket
val dataStream = env.socketTextStream("localhost", port, '\n')



import org.apache.flink.streaming.api.scala._
val dataStreamDeal = dataStream.flatMap( w => w.split("\\s") ).map( w => 
WordWithCount(w,1))
  .keyBy("word")
  //将当前window中所有的行记录，发送过来ProcessAllWindowFunction函数中去处理(可以排序，可以对相同key进行处理)
  //缺点，window中数据量大时，就容易内存溢出
  .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))

  .process(new 
ProcessAllWindowFunction[WordWithCount,WordWithCount,TimeWindow] {
override def process(context: Context, elements: 
Iterable[WordWithCount], out: Collector[WordWithCount]): Unit = {
  val set = new mutable.HashSet[WordWithCount]{}


  for(wordCount <- elements){
if(set.contains(wordCount)){
  set.remove(wordCount)
  set.add(new WordWithCount(wordCount.word,wordCount.count + 1))
}else{
  set.add(wordCount)
}
  }

  val sortSet = set.toList.sortWith( (a,b) => a.word.compareTo(b.word)  
< 0 )

  for(wordCount <- sortSet)  out.collect(wordCount)
}

  })




  //.countWindow(3)
  //.countWindow(3,1)
  //.countWindowAll(3)




//textResult.print().setParallelism(1)

val bucketingSink = new 
BucketingSink[WordWithCount]("file:/opt/n_001_workspaces/bigdata/flink/flink-maven-scala-2/sink-data")


bucketingSink.setBucketer(new 
DateTimeBucketer[WordWithCount]("-MM-dd--HHmm", ZoneId.of("Asia/Shanghai")))
//bucketingSink.setWriter(new SequenceFileWriter[IntWritable, Text]())
//bucketingSink.setWriter(new SequenceFileWriter[WordWithCount]())
//bucketingSink.setBatchSize(1024 * 1024 * 400) // this is 400 MB,
//bucketingSink.setBatchSize(100 ) // this is 400 MB,
bucketingSink.setBatchSize(1024 * 1024 * 400 ) // this is 400 MB,
//bucketingSink.setBatchRolloverInterval(20 * 60 * 1000); // this is 20 mins
bucketingSink.setBatchRolloverInterval( 2 * 1000); // this is 20 mins
//setInactiveBucketCheckInterval
//setInactiveBucketThreshold
//每间隔多久时间，往Sink中写数据，不是每天条数据就写，浪费资源

bucketingSink.setInactiveBucketThreshold(2 * 1000)
bucketingSink.setAsyncTimeout(1 * 1000)


dataStreamDeal.setParallelism(1)
  .addSink(bucketingSink)




if(args == null || args.size ==0){
  env.execute("默认作业")

  //执行计划
  //println(env.getExecutionPlan)
  //StreamGraph
 //println(env.getStreamGraph.getStreamingPlanAsJSON)



  //JsonPlanGenerator.generatePlan(jobGraph)

}else{
  env.execute(args(0))
}

println("结束")

  }


  // Data type for words with count
  case class WordWithCount(word: String, count: Long)

/*  abstract private   class

????????????????????????????????????????

2019-03-04 文章 ??????


stream api??
!

Re: [问题]Flink并行计算中，不同的Window是如何接收到自己分区的数据的，即Window是如何确定当前Window属于哪个分区数?

2019-03-04 文章刘文


感谢大家的回答，我明白了一些了，并整理这个问题的文档

Flink1.7.2 Source、Window数据交互源码分析: 
https://github.com/opensourceteams/fink-maven-scala-2/blob/master/md/miniCluster/flink-source-window-data-exchange.md
 

Flink1.7.2 并行计算源码分析: 
https://github.com/opensourceteams/fink-maven-scala-2/blob/master/md/miniCluster/Flink-Parallelism-Calculation.md
 







Flink1.7.2 并行计算源码分析

 
源码

源码:https://github.com/opensourceteams/fink-maven-scala-2 

Flink1.7.2 Source、Window数据交互源码分析: 
https://github.com/opensourceteams/fink-maven-scala-2/blob/master/md/miniCluster/flink-source-window-data-exchange.md
 

 
概述

Flink Window如何进行并行计算
Flink source如何按key,hash分区，并发射到对应分区的下游Window
 
WordCount
 程序

package 
com.opensourceteams.module.bigdata.flink.example.stream.worldcount.nc.parallelism

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * nc -lk 1234  输入数据
  */
object SocketWindowWordCountLocal {



  def main(args: Array[String]): Unit = {


val port = 1234
// get the execution environment
   // val env: StreamExecutionEnvironment = 
StreamExecutionEnvironment.getExecutionEnvironment


val configuration : Configuration = getConfiguration(true)

val env:StreamExecutionEnvironment = 
StreamExecutionEnvironment.createLocalEnvironment(1,configuration)





// get input data by connecting to the socket
val dataStream = env.socketTextStream("localhost", port, '\n')



import org.apache.flink.streaming.api.scala._
val textResult = dataStream.flatMap( w => w.split("\\s") ).map( w => 
WordWithCount(w,1))
  .keyBy("word")
  /**
* 每20秒刷新一次，相当于重新开始计数，
* 好处，不需要一直拿所有的数据统计
* 只需要在指定时间间隔内的增量数据，减少了数据规模
*/
  .timeWindow(Time.seconds(5))
  //.countWindow(3)
  //.countWindow(3,1)
  //.countWindowAll(3)


  .sum("count" )

textResult
  .setParallelism(3)
  .print()




if(args == null || args.size ==0){


  
println("==以下为执行计划==")
  println("执行地址(firefox效果更好):https://flink.apache.org/visualizer;)
  //执行计划
  //println(env.getExecutionPlan)
 // println("==以上为执行计划 
JSON串==\n")
  //StreamGraph
 println(env.getStreamGraph.getStreamingPlanAsJSON)



  //JsonPlanGenerator.generatePlan(jobGraph)

  env.execute("默认作业")

}else{
  env.execute(args(0))
}

println("结束")

  }


  // Data type for words with count
  case class WordWithCount(word: String, count: Long){
//override def toString: String = Thread.currentThread().getName + word + " 
: " + count
  }


  def getConfiguration(isDebug:Boolean = false):Configuration = {

val configuration : Configuration = new Configuration()

if(isDebug){
  val timeout = "10 s"
  val timeoutHeartbeatPause = "100 s"
  configuration.setString("akka.ask.timeout",timeout)
  configuration.setString("akka.lookup.timeout",timeout)
  configuration.setString("akka.tcp.timeout",timeout)
  configuration.setString("akka.transport.heartbeat.interval",timeout)
  
configuration.setString("akka.transport.heartbeat.pause",timeoutHeartbeatPause)
  configuration.setString("akka.watch.heartbeat.pause",timeout)
  configuration.setInteger("heartbeat.interval",1000)
  configuration.setInteger("heartbeat.timeout",5000)
}


configuration
  }


}


 
输入数据

1 2 3 4 5 6 7 8 9 10
 
源码分析

回复：jobmanager重启之后，job没有重启

2019-03-04 文章 Fei Han

杨森：
   你好！对于Flink on yarn 模式 flink-conf.yaml中的配置如下：

 high-availability: zookeeper
 high-availability.zookeeper.quorum: cdh1:2181,cdh2:2181,cdh3:2181
 high-availability.storageDir: hdfs://cdh1:9000/flink/recovery
 high-availability.zookeeper.path.root: /flink
 state.backend: filesystem
 state.backend.fs.checkpointdir: hdfs://cdh1:9000/flink/checkpoints
 fs.hdfs.hadoopconf: /usr/local/hadoop-2.7.4/etc/hadoop
yarn-site.xml中的配置如下：

 
 yarn.resourcemanager.am.max-attempts
 4
 
以上的默认配置，你也可以在程序中配置，默认配置和程序都配置试试。

如图看到Attempt ID由01变为02，进程也有所变化，，说明HA切换成功了。
 希望这些内容帮助到你。
  Best,
  韩非


--
发件人：孙森 
发送时间：2019年3月4日(星期一) 18:41
收件人：user-zh 
主　题：jobmanager重启之后，job没有重启

大家好：
  有配置过flink ha吗？我配置了yarn高可用之后，kill掉job manager 
之后，jobmanager可以成功重启，但是yarn 
session上正在运行的job并没有重启。我程序是开启了checkpoint的。有谁遇到过吗？是我配置的问题还是什么原因？

附上高可用配置：

high-availability.storageDir: hdfs:///flink/ha/
high-availability.zookeeper.quorum: h1:2181,h2:2181,h3:2181
high-availability.zookeeper.path.root: /flink
yarn.application-attempts: 2


Thanks!

jobmanager重启之后，job没有重启

2019-03-04 文章孙森

大家好：
  有配置过flink ha吗？我配置了yarn高可用之后，kill掉job manager 
之后，jobmanager可以成功重启，但是yarn 
session上正在运行的job并没有重启。我程序是开启了checkpoint的。有谁遇到过吗？是我配置的问题还是什么原因？

附上高可用配置：

high-availability.storageDir: hdfs:///flink/ha/
high-availability.zookeeper.quorum: h1:2181,h2:2181,h3:2181
high-availability.zookeeper.path.root: /flink
yarn.application-attempts: 2


Thanks!

Re:回复：sql-client 支持远程连接flink集群吗

2019-03-04 文章 Yuan Yifan

请教一下，如何使用 Zeppelin 编写 FlinkSQL并且执行看结果呢（就像SQLClient那样），我找了一些资料，只找到使用Scala的：
https://zeppelin.apache.org/docs/0.8.0/interpreter/flink.html


顺便问一下，我现在已经写了一些代码用来初始化 StreamEnvironment 以及 TableEnvironment，我如何将这些初始化的代码用到 
Flink SQL Client 中呢（而不是使用配置文件初始化表环境）？我尝试在Flink的工程中引入 Flink SQL 
Client的maven包，并且以 minimal cluster 模式启动，但是无法执行。



在 2019-03-04 10:34:27，"Fei Han"  写道：
>你好！
>sql-client你现在可以在zeepelin上去写SQL，但只支持DML，DDL暂时
>  还不支持(如 create table)
>
>Best，
>  Han Fei
>
>
>
>
>--
>发件人：yuess_coder <642969...@qq.com>
>发送时间：2019年3月4日(星期一) 10:29
>收件人：user-zh@flink.apache.org 
>主　题：sql-client 支持远程连接flink集群吗
>
>如题
>

Re: 订阅

2019-03-04 文章刘文

请发送邮件到 :  user-zh-subscr...@flink.apache.org 


回收到一份回执邮件，有可能在垃圾邮件中，你再回复任意消息，就可以订阅了






> 在 2019年3月2日，下午3:55，(｡･ˇ_ˇ･｡:)  写道：
> 
> 订阅

Re: flink学习

2019-03-04 文章刘文




请发送邮件到 :  user-zh-subscr...@flink.apache.org

回收到一份回执邮件，有可能在垃圾邮件中，你再回复任意消息，就可以订阅了


> 在 2019年3月2日，下午3:10，袁刚  写道：
> 
> |
> flink学习
> 
> 
> 
> 
> 
> 
> 
> 
> 
> 
> |
> |
> |   |
> 
> |
> |

回复: Re: [问题]Flink并行计算中，不同的Window是如何接收到自己分区的数据的，即Window是如何确定当前Window属于哪个分区数?

2019-03-04 文章 343122...@qq.com

以下个人理解,可能不100%准确.
是根据keyBy,即你代码中的 .keyBy("word"), 
根据其值的hash值,模并行度得到余数, 来确定 数据该分到哪个分区, 
你代码里没有指定时间特征,默认是处理时间.
所有你的window,则是根据处理时间来分窗口的.




343122...@qq.com
 
发件人： 刘 文
发送时间： 2019-03-04 11:53
收件人： user-zh@flink.apache.org
主题： Re: [问题]Flink并行计算中，不同的Window是如何接收到自己分区的数据的，即Window是如何确定当前Window属于哪个分区数?
--
很抱歉，我还是没有理解，我可以再次请求帮助吗？
 
例如：
).并行度调置为2时setParallelism(2)，会产生两个window线程
). 流 WordCount local ,flink 1.7.2
).这两个Window线程是如何读取到自己分区中的数据的，Window分区是如何确定的?
).输入数据
  1 2 3 4 5 6 7 8 9 10
).source   ->  operator   -> 
--
change [partition 0]
   
   
key:1partition:0
key:2partition:0
key:3partition:0
key:4partition:0
key:6partition:0
key:10   partition:0
--
change 1  [partition 1]
key:5partition:1
key:7partition:1
key:8partition:1
key:9partition:1
).window 0 （1/2）
window 当前partition是如何确定的?
window 是如何读到当前parition中的数据的?
   
).window 1 （2/2） 
window 当前partition是如何确定的?
window 是如何读到当前parition中的数据的?
 
 
--
 
 
 
 
 
> 在 2019年3月3日，下午9:26，刘 文  写道：
> 
> WordCount.scala
> package 
> com.opensourceteams.module.bigdata.flink.example.stream.worldcount.nc.parallelism
> 
> import org.apache.flink.configuration.Configuration
> import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
> import org.apache.flink.streaming.api.windowing.time.Time
> 
> /**
>  * nc -lk 1234  输入数据
>  */
> object SocketWindowWordCountLocal {
> 
> 
> 
>  def main(args: Array[String]): Unit = {
> 
> 
>val port = 1234
>// get the execution environment
>   // val env: StreamExecutionEnvironment = 
> StreamExecutionEnvironment.getExecutionEnvironment
> 
> 
>val configuration : Configuration = getConfiguration(true)
> 
>val env:StreamExecutionEnvironment = 
> StreamExecutionEnvironment.createLocalEnvironment(1,configuration)
> 
> 
> 
> 
> 
>// get input data by connecting to the socket
>val dataStream = env.socketTextStream("localhost", port, '\n')
> 
> 
> 
>import org.apache.flink.streaming.api.scala._
>val textResult = dataStream.flatMap( w => w.split("\\s") ).map( w => 
> WordWithCount(w,1))
>  .keyBy("word")
>  /**
>* 每20秒刷新一次，相当于重新开始计数，
>* 好处，不需要一直拿所有的数据统计
>* 只需要在指定时间间隔内的增量数据，减少了数据规模
>*/
>  .timeWindow(Time.seconds(5))
>  //.countWindow(3)
>  //.countWindow(3,1)
>  //.countWindowAll(3)
> 
> 
>  .sum("count" )
> 
>textResult
>  .setParallelism(100)
>  .print()
> 
> 
> 
> 
>if(args == null || args.size ==0){
> 
> 
>  
> println("==以下为执行计划==")
>  println("执行地址(firefox效果更好):https://flink.apache.org/visualizer;)
>  //执行计划
>  println(env.getExecutionPlan)
>  println("==以上为执行计划 
> JSON串==\n")
>  //StreamGraph
> //println(env.getStreamGraph.getStreamingPlanAsJSON)
> 
> 
> 
>  //JsonPlanGenerator.generatePlan(jobGraph)
> 
>  env.execute("默认作业")
> 
>}else{
>  env.execute(args(0))
>}
> 
>println("结束")
> 
>  }
> 
> 
>  // Data type for words with count
>  case class WordWithCount(word: String, count: Long)
> 
> 
>  def getConfiguration(isDebug:Boolean = false):Configuration = {
> 
>val configuration : Configuration = new Configuration()
> 
>if(isDebug){
>  val timeout = "10 s"
>  val timeoutHeartbeatPause = "100 s"
>  configuration.setString("akka.ask.timeout",timeout)
>  configuration.setString("akka.lookup.timeout",timeout)
>  configuration.setString("akka.tcp.timeout",timeout)
>  configuration.setString("akka.transport.heartbeat.interval",timeout)
>  
> configuration.setString("akka.transport.heartbeat.pause",timeoutHeartbeatPause)
>  configuration.setString("akka.watch.heartbeat.pause",timeout)
>  configuration.setInteger("heartbeat.interval",1000)
>  configuration.setInteger("heartbeat.timeout",5000)
>}
> 
> 
>configuration
>  }
> 
> 
> }
> 
> 
> 
>> 在 2019年3月3日，下午9:05，刘 文  写道：
>> 
>> 
> [问题]Flink并行计算中，不同的Window是如何接收到自己分区的数据的，即Window是如何确定当前Window属于哪个分区数?
>> 
>> ).环境 Flink1.7.2 WordCount local,流处理
>> ).source 中 RecordWriter.emit(),给每个元素按key,分到不同的partition,已确定每个元素的分区位置,分区个数由 
>> DataStream.setParallelism(2)决定
>> 
>> public void emit(T record) throws IOException, InterruptedException {
>>emit(record, channelSelector.selectChannels(record, numChannels));
>> }
>> 
>> 通过copyFromSerializerToTargetChannel(int targetChannel) 
>> 往不同的通道写数据，就是往不同的分区对应的window发送数据(数据是一条一条发送)
>> ).有多少个并行度，DataStream.setParallelism(2) ，就开启多少个Window
>> 
>

flink学习

2019-03-04 文章袁刚

|
flink学习










|
|
|   |

|
|

Re: jobmanager重启之后，job没有重启

Re: 如何每五分钟统计一次当天某个消息的总条数

?????? ????????????????????????????????????????

Re: 如何每五分钟统计一次当天某个消息的总条数

Re: 如何每五分钟统计一次当天某个消息的总条数

????????????????????????????????????????

Re: [问题]Flink并行计算中，不同的Window是如何接收到自己分区的数据的，即Window是如何确定当前Window属于哪个分区数?

回复：jobmanager重启之后，job没有重启

jobmanager重启之后，job没有重启

Re:回复：sql-client 支持远程连接flink集群吗

Re: 订阅

Re: flink学习

回复: Re: [问题]Flink并行计算中，不同的Window是如何接收到自己分区的数据的，即Window是如何确定当前Window属于哪个分区数?

flink学习

14 matches

Site Navigation

Mail list logo

Footer information