Flink批连接性能

Question

我一直在批处理模式下测试使用TableApi和DataStream api的简单联接。然而，我得到了非常糟糕的结果，所以它一定是我做错了什么。用于联接的数据集约为 900gb 和 3gb。用于测试的环境是具有 10 * m5.xlarge 工作节点的 EMR。

使用的TableApi方法是在数据s3路径上创建一个表，并在目标s3路径上对创建的表执行插入语句。通过调整任务管理器内存、numberOfTaskSlot、并行性，但无法使其在某种程度上可以接受的时间内执行（至少1.5h）。

在批处理模式下使用DataStreamApi时，我总是会遇到这样一个问题：由于Thread占用了超过90%的磁盘空间，因此它会终止任务。所以我不知道这是因为代码，还是仅仅因为flink需要比spark更多的磁盘空间。读取数据流：

    val sourceStream: DataStream[SourceObj] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "someSourceName")
      .map(x => SourceObj.convertFromString(x))

加入：

    val joinedStream = sourceStream.join(sourceStream2)
      .where(col1 => sourceStream.col1)
      .equalTo(col2 => sourceStream2.col2)
      .window(GlobalWindows.create())
      .trigger(CountTrigger.of(1))
      .apply{
        (s, c) => JoinedObj(c.col1, s.col2, s.col3)
      }

是我缺少某些内容，还是只需要纵向扩展群集？

Answer 1

当您在批处理模式下使用DataStream API时，它会在所有随机/加入/减少操作中广泛使用托管内存。此外，如最后一段所述，Flink将在加入期间将无法放入内存的所有数据溢出到磁盘。

因此，我认为这可能是磁盘空间不足问题的原因。我的工作也遇到了同样的问题。

Answer 2

一般来说，您最好使用Flink的Table/SQLAPI实现关系工作负载，这样它的优化器就有机会提供帮助。

但是如果我没有理解错的话，这个特殊的连接执行起来会非常昂贵，因为没有任何东西会从状态中过期。这两个表都将在Flink中完全具体化，因为对于这个查询，每一行输入都是相关的，并可能影响结果。

如果您可以将它转换为具有时间约束的某种连接，优化程序可以使用该约束来释放不再有用的行，那么它的行为将会好得多。