Flink löschen alte Prüfpunktdatei

Question

Wir verwenden Dateisystem als Prüfpunktstatus-Backend. unsere Einstellung finden:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
env.setStreamTimeCharacteristic(timeCharacteristic); 
env.getConfig().setAutoWatermarkInterval(Time.seconds(30).toMilliseconds()); 
env.enableCheckpointing(Time.seconds(30).toMilliseconds()); 
CheckpointConfig checkpointConfig = env.getCheckpointConfig(); 
checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); 
env.setStateBackend(new FsStateBackend(props.getProperty("backendstate.path"))); 

Dies funktioniert während der Entwicklung in Ordnung, aber sobald wir die Produktion bereitstellen, fanden wir heraus, dass die Checkpoint-Dateien schnell Speicherplatz fressen. Ich habe den Befehl ls einem der Checkpoint-Verzeichnis und finden, dass

chk-10 chk-104 chk-13 chk-18 chk-23 chk-28 chk-32 chk-37 chk-41 chk-46 chk-50 chk-55 chk-6 chk-64 chk-69 chk-73 chk-78 chk-82 chk-87 chk-91 chk-96 chk-100 ... hundreds of them 

Ich frage mich, ob es einen Weg gibt, die älteren chk-x-Dateien zu löschen oder wir haben sie manuell entfernen?

Answer 1

Sie könnten versuchen, diese

CheckpointConfig config = env.getCheckpointConfig(); 
config.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); 

Der ExternalizedCheckpointCleanup Modus konfiguriert, was mit externalisierte Checkpoints passiert, wenn Sie den Auftrag abbrechen:

ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION: Bewahren Sie die externalisiert Kontrollpunkt, wenn der Job ist abgebrochen. Beachten Sie, dass Sie in diesem Fall den Prüfpunktstatus nach der Stornierung manuell bereinigen müssen.

ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: die externalisiert Kontrollpunkt löschen, wenn der Auftrag abgebrochen wird. Der Checkpoint-Status ist nur verfügbar, wenn der Job fehlschlägt.

Answer 2

Während des Einsatzes legen Sie die folgende Eigenschaft in der „conf/flink-conf.yaml“ Datei auf die maximale Anzahl von Checkpoints Sie behalten möchten:

state.checkpoints.num-retained 

Idealerweise sollte es mehr als 1 sein , da es bei der Wiederherstellung des Jobs hilft, wenn der letzte Checkpoint beschädigt ist.

Weitere Informationen finden Sie in diesem Artikel: Config Options