2

我想加入一个大表，不可能包含在TM内存和流（kakfa）中。我成功地加入了我的测试，将 table-api 与 datastream api 混合在一起。我做了以下事情：

val stream: DataStream[MyEvent] = env.addSource(...)
stream
   .timeWindowAll(...)
   .trigger(...)
   .process(new ProcessAllWindowFunction[MyEvent, MyEvent, TimeWindow] {
        
        var tableEnv: StreamTableEnvironment = _
        
        override def open(parameters: Configuration): Unit = {
          //init table env
        }

        override def process(context: Context, elements: Iterable[MyEvent], out: Collector[MyEvent]): Unit = {
          val table = tableEnv.sqlQuery(...)
          elements.map(e => {
            //do process
            out.collect(...)
          })
        }
      })

它正在工作，但我从未在任何地方看到过这种类型的实现。可以吗？有什么缺点？

4

1 回答 1

1

不应该在 Flink 函数中使用StreamExecutionEnvironment或使用。TableEnvironment环境用于构建提交给集群的管道。

您的示例在集群作业中向集群提交作业。

这可能适用于某些用例，但通常不鼓励使用。想象一下，您的外部流包含数千个事件，您的函数将为每个事件创建一个作业，它可能对您的集群进行 DDoS。

于 2021-03-12T09:19:04.487 回答