Flink应用的逻辑执行图

2021-07-11 11:00:23 阅读：169 来源： 互联网

标签：逻辑 keyBy word val Flink 并行度应用算子 port

这里以最基础的wordcount程序说明flink应用的逻辑执行图，程序代码如下：

object SetParalWC {
  def main(args: Array[String]): Unit = {
    //创建流处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //    env.setParallelism(8) 设置所有算子任务并行度为8，同时也可以每个算子单独设置该参数


    // 从程序运行参数中读取hostname和port
    val params: ParameterTool = ParameterTool.fromArgs(args)
    val hostname: String = params.get("host")
    val port: Int = params.getInt("port")
    // 接受socket文本流，nc -lk 8888
    val inputDataStream: DataStream[String] = env.socketTextStream(hostname, port)

    // 定义转换操作，word count
    val resultDataStream: DataStream[(String, Int)] = inputDataStream
      .flatMap(_.split(" "))   // 以空格分词，打散得到所有的word
      .filter(_.nonEmpty)
      .map( (_, 1) )  // 转换成(word, count)二元组
      .keyBy(0)  // 按照第一个元素分组
      .sum(1)  // 按照第二个元素求和，这里就能体现出有状态的流处理，结果是累计滚动输出的

    resultDataStream.print()
    // 如果在本地运行任务没有设定并行度，那么并行度默认与核数一致
    /**
     * 3> (world,1)
     * 2> (hello,1)
     * 4> (flink,1)
     * 2> (hello,2)
     * 1> (spark,1)
     * 2> (hello,3)
     * 3> (future,1)
     * 4> (stream,1)
     * 4> (is,1)
     * 4> (why,1)
     * 这里的数字代表的就是数据keyBy后的子任务序号
     */
    // resultDataStream.print().setParallelism(1)
    env.execute("stream word count job")
  }
}

该应用通过WebUI查看应用逻辑执行图如下：

可以看到程序通过source算子获取数据的并行度是1，但后面任务算子的任务并行度为2，所以数据需要经过rebalance才能分配到后续任务中，同时注意到任务链的存在。

keyBy算子在逻辑执行图中并没有体现出来，在任务链之后通过HASH的方式直接将数据传输到了aggregation操作中，而aggregation操作对应的应该是sum()算子，这说明keyBy不是计算方式，而是一种定义数据传输方式的算子。

标签：逻辑,keyBy,word,val,Flink,并行度,应用,算子,port
来源： https://www.cnblogs.com/nineTong/p/14998134.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Flink应用的逻辑执行图