与flink并行性相关的对象实例&应用方法

Question

>

先让我问一下我的问题，然后你能澄清一下我对应用方法的设想吗？

问题：如果我的应用程序在每一分钟的间隔内创建1.500.000（大约）条记录，并且flink job使用15++不同的运算符从kafka consumer读取这些记录，那么这个逻辑可能会产生延迟、背压等？（可以假设并行度为16）

public class Sample{
  //op1 = 
     kafkaSource
                .keyBy(something)
                .timeWindow(Time.minutes(1))
                .apply(new ApplySomething())
                .name("Name")
                          .addSink(kafkaSink);
  //op2 = 
    kafkaSource
                .keyBy(something2)
                .timeWindow(Time.seconds(1)) // let's assume that this one second
                .apply(new ApplySomething2())
                .name("Name")
                          .addSink(kafkaSink);
 // ...

  //op16 = 
    kafkaSource
                .keyBy(something16)
                .timeWindow(Time.minutes(1)) 
                .apply(new ApplySomething16())
                .name("Name")
                          .addSink(kafkaSink);

}
// ..
public class ApplySomething ... {
  private AnyObject object;
  private int threshold = 30, 40, 100 ...;

      @Override
    public void open(Configuration parameters) throws Exception{
        object = new AnyObject();
    }

    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<Record> input, Collector<Result> out) throws Exception{
        int counter = 0;
        for (Record each : input){
          counter += each.getValue();
          if (counter > threshold){
            out.collec(each.getResult());
            return;
          }
        }
    }
}

• 如果是，是否应该使用带有状态的flatMap（rocksDB）而不是TimeWindow？
• 我的预测是“是”。让我解释一下为什么我会这样想：
  • 如果并行度为16，则将有16个不同的indivudualapplySomething1()、ApplySomething2()...applySomething16()实例，并且每个applySomething..()类将有16个AnyObject()实例。
  • 应用程序工作时，如果keyby(something)分区数大于16（假设我的应用程序每天有1.00.000个不同的something)，则一些applysomething..()实例将处理不同的键，因此一个apply()应该在处理之前等待其他实例循环。那么这会造成延迟吗？

Answer 1

闪烁的时间窗口与历元对齐（例如，如果你有一堆小时窗口，它们都会在小时触发）。因此，如果你真的打算在你的工作中有一堆不同的窗口，你应该配置它们有不同的偏移量，这样它们就不会同时被触发。那样做会分散负荷。看起来像这样

.window(TumblingProcessingTimeWindows.of(Time.minutes(1), Time.seconds(15))

（或视情况使用TumblingEventTimeWindows)。这将创建一分钟长的窗口，每分钟后15秒触发。

只要用例允许，就应该使用增量聚合（通过reduce或aggrege)，而不是使用WindowFunction（或ProcessWindowFunction)，后者必须收集分配给列表中每个窗口的所有事件，然后将它们作为一种小型批处理进行处理。

键控时间窗口将保持其状态在RocksDB中，假设您已经将RocksDB配置为您的状态后端。您不需要切换到使用richflatmap来访问rocksdb。（此外，由于flatMap不能使用计时器，我认为您最终会使用process函数。）

当window操作符的任何并行实例忙于执行它的窗口函数(applysomethings)时，您认为该任务不会执行任何其他操作是正确的--因此它将（除非非常快地完成）产生暂时的背压。您将希望根据需要增加并行度，以便作业能够满足您对吞吐量和延迟的要求。