首页 文章

Flink:一个程序中的Dataset和Datastream API . 可能吗?

提问于
浏览
2

我想首先使用数据集API操作静态数据,然后使用DataStream API来运行流式作业 . 如果我在IDE上编写代码,它就能完美运行 . 但是当我尝试在本地flink jobmanager(所有并行1)上运行时,流代码永远不会执行!

例如,以下代码不起作用:

val parallelism = 1

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(parallelism)

val envStatic = ExecutionEnvironment.getExecutionEnvironment
envStatic.setParallelism(parallelism)

val myStaticData = envStatic.fromCollection(1 to 10)
val myVal: Int = myStaticData.reduce(_ + _).collect().head

val theStream = env.fromElements(1).iterate( iteretion => {
  val result = iteretion.map(x => x + myVal)
  (result, result)
})
theStream.print()
env.execute("static and streaming together")

我应该尝试让这个东西工作?

日志:execution logs for above program

执行计划:plan似乎是一个循环 .

1 回答

  • 4

    如果您有一个由多个子作业组成的Flink作业,例如由 countcollectprint 触发,则无法通过Web界面提交作业 . Web界面仅支持单个Flink作业 .

相关问题