Jaming R 2020-04-01 17:00 采纳率: 0%
浏览 444

flink 并行度导致输出控制台结果问题

我的电脑是8线程,当我运行下面这段代码时,可以输出结果,但是当把并行度注释掉时,就不输出结果了,这是为什么呢?输入数据一样,watermark应该都是一样的啊

输入数据为:
1585721697000,xiao,8
1585721700000,xiao,10
1585721705000,xiao,4
1585721715000,xiao,9

case class Student(id:Long,name:String,age:Int)
object test3 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(4)
    //指定时间类型
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val value: DataStream[String] = env.socketTextStream("localhost", 9999)

    val map: DataStream[Student] = value.map(line => {
      val arr: Array[String] = line.split(",")
      Student(arr(0).toLong, arr(1), arr(2).toInt)
    })
    val value1: DataStream[Student] = map.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Student](Time.seconds(3)) {
      override def extractTimestamp(t: Student): Long = {
        t.id
      }
    })
    val value2: KeyedStream[Student, String] = value1.keyBy(_.name)
    val value3: WindowedStream[Student, String, TimeWindow] = value2.timeWindow(Time.seconds(10))
    val value4: DataStream[Student] = value3.sum(2)
    value4.print()
    env.execute()
  }
}

  • 写回答

0条回答 默认 最新

    报告相同问题?

    悬赏问题

    • ¥35 平滑拟合曲线该如何生成
    • ¥100 c语言,请帮蒟蒻写一个题的范例作参考
    • ¥15 名为“Product”的列已属于此 DataTable
    • ¥15 安卓adb backup备份应用数据失败
    • ¥15 eclipse运行项目时遇到的问题
    • ¥15 关于#c##的问题:最近需要用CAT工具Trados进行一些开发
    • ¥15 南大pa1 小游戏没有界面,并且报了如下错误,尝试过换显卡驱动,但是好像不行
    • ¥15 自己瞎改改,结果现在又运行不了了
    • ¥15 链式存储应该如何解决
    • ¥15 没有证书,nginx怎么反向代理到只能接受https的公网网站