«Буферный пул уничтожен», когда я использую Flink SlidingEventTimeWindows - PullRequest
Новая
       7

«Буферный пул уничтожен», когда я использую Flink SlidingEventTimeWindows

0 голосов
/ 15 ноября 2018

Flink выбрасывает «java.lang.IllegalStateException: буферный пул уничтожается», когда я использую «SlidingEventTimeWindows», но все становится нормально, когда я меняю на «SlidingProcessingTimeWindows».

Трассировка стека следующая: 

18:37:53,728 WARN  org.apache.flink.streaming.api.operators.AbstractStreamOperator  - Error while emitting latency marker.
java.lang.RuntimeException: Buffer pool is destroyed.
	at org.apache.flink.streaming.runtime.io.RecordWriterOutput.emitLatencyMarker(RecordWriterOutput.java:147)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.emitLatencyMarker(AbstractStreamOperator.java:683)
	at org.apache.flink.streaming.api.operators.StreamSource$LatencyMarksEmitter$1.onProcessingTime(StreamSource.java:151)
	at org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$RepeatedTriggerTask.run(SystemProcessingTimeService.java:330)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalStateException: Buffer pool is destroyed.
	at org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestMemorySegment(LocalBufferPool.java:230)
	at org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestBufferBuilderBlocking(LocalBufferPool.java:204)
	at org.apache.flink.runtime.io.network.api.writer.RecordWriter.requestNewBufferBuilder(RecordWriter.java:213)
	at org.apache.flink.runtime.io.network.api.writer.RecordWriter.sendToTarget(RecordWriter.java:144)
	at org.apache.flink.runtime.io.network.api.writer.RecordWriter.randomEmit(RecordWriter.java:125)
	at org.apache.flink.streaming.runtime.io.StreamRecordWriter.randomEmit(StreamRecordWriter.java:93)
	at org.apache.flink.streaming.runtime.io.RecordWriterOutput.emitLatencyMarker(RecordWriterOutput.java:144)
	... 10 more

Я наконец решил следующие шаги.

Сначала замените «collect» на «collectWithTimestamp» в My DataMockSource, который используется для генерации потоковых данных. После этого в консоли исчезнет «Ошибка при отправке маркера задержки».

Во-вторых, замените BoundedOutOfOrdernessTimestampExtractor на AscendingTimestampExtractor, который используется для обработки EventTime. В моем DataMockSource я генерирую данные и отправляю их одновременно, поэтому AscendingTimestampExtractor является правильным способом создания водяного знака.

Я публикую здесь основной код и полный проект на github . Надеюсь, это полезно. 

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(10000); //

DataStreamSource<MockData> mockDataDataStreamSource = env.addSource(new DataMockSource());
mockDataDataStreamSource.assignTimestampsAndWatermarks(
    new AscendingTimestampExtractor<MockData>() {
      @Override
      public long extractAscendingTimestamp(MockData element) {
        return element.getTimestamp();
      }
    });

SingleOutputStreamOperator<Tuple2<String, Long>> countStream = mockDataDataStreamSource
    .keyBy("country").window(
        SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(10)))
//        .allowedLateness(Time.seconds(5))
    .process(
        new FlinkEventTimeCountFunction()).name("count elements");

countStream.addSink(new SinkFunction<Tuple2<String, Long>>() {
  @Override
  public void invoke(Tuple2<String, Long> value, Context context) throws Exception {
    System.out.println(value);
  }
});

env.execute("count test ");

Мой DataMockSource находится здесь: 

private volatile boolean running = true;
  @Override
  public void run(SourceContext sourceContext) throws Exception {
    while (running){
      MockData mockData = new MockData();
      mockData.setAge(ThreadLocalRandom.current().nextInt(1,99));
      mockData.setCountry("country "+ThreadLocalRandom.current().nextInt(2,5));
      mockData.setId(ThreadLocalRandom.current().nextLong());
      mockData.setTimestamp(Instant.now().toEpochMilli());
      // emit record with timestamp
      sourceContext.collectWithTimestamp(mockData,Instant.now().toEpochMilli());
//      sourceContext.collect(mockData);

      TimeUnit.SECONDS.sleep(3);
    }
  }

  @Override
  public void cancel() {
     running = false;
  }

1 Ответ

0 голосов
/ 15 ноября 2018

При работе во время события вам необходимо организовать извлечение метки времени и создание водяных знаков, либо в вашем источнике, либо с помощью assignTimestampsAndWatermarks. Похоже, вы этого не делаете, что объясняет, почему вы не получите никакого вывода (окно времени-события никогда не сработает).

Кроме того, ваш источник должен иметь метод отмены. Примерно так: 

private volatile boolean running = true;

@Override
public void run(SourceContext ctx) throws Exception {
    while (running) {
        ...
    }
}

@Override
public void cancel() {
    running = false;
}

Я думаю, это может объяснить исключение, которое вы видите. Возможно, источник продолжает работать и отправлять маркеры задержки после того, как задание начало закрываться.

...