如何应对外部检查点在弗林克中表现出来

Question

我使用的是带有岩石的增量检查点的Flink 1.4.2，并将检查点保存到S3存储桶中。检查点的结构是一个清单文件，指向某些包含状态的文件。当我在文本编辑器中打开清单文件时，我会看到一些不可读的块和一些S3 URL。

我如何应此清单文件以获取S3 URL的列表？

Answer 1

https://github.com/king/bravo and https://github.com/sjwiesman/flink/tree/savepoint-connector包含可以读取和写入SavePoints/savepoints/savepoints/检查点。您可能会发现一个或两个有用，无论是直接或示例。请参阅此JIRA票-https：//issues.apache.org/jira/jira/browse/flink-12047-遵循正在进行的工作，以创建更好的工具，以使用Flink快照。

Answer 2

类SavepointStore表格Apache Flink Runtime库包含存储和加载保存点的方法。

仅针对我当前的情况，我创建了此片段来检索与检查点有关的文件。

import org.apache.flink.runtime.checkpoint.savepoint.Savepoint;
import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore;
import org.apache.flink.runtime.state.IncrementalKeyedStateHandle;
import org.apache.flink.runtime.state.KeyGroupsStateHandle;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.filesystem.FileStateHandle;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
import java.io.IOException;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class CheckpointFileLocator {
    public static void main(String[] args) throws IOException {
        System.out.println(new CheckpointFileLocator()
                .getS3Locations("/Users/ezequiel/Downloads/chk-3-checkpoint_metadata-f350e54becb2"));
    }
    public Set<String> getS3Locations(String manifestPath) throws IOException {
        Savepoint savepoint = SavepointStore.loadSavepoint(manifestPath, this.getClass().getClassLoader());
        Stream<String> rawStream = savepoint.getOperatorStates().stream()
                .flatMap(operatorState -> operatorState.getSubtaskStates().values().stream())
                .flatMap(operatorSubtaskState -> operatorSubtaskState.getRawKeyedState().stream())
                .map(keyedStateHandle -> (KeyGroupsStateHandle) keyedStateHandle)
                .map(KeyGroupsStateHandle::getDelegateStateHandle)
                .map(this::getPath);
        Stream<String> metadataStream = savepoint.getOperatorStates().stream()
                .flatMap(operatorState -> operatorState.getSubtaskStates().values().stream())
                .flatMap(operatorSubtaskState -> operatorSubtaskState.getManagedKeyedState().stream())
                .map(keyedStateHandle -> (IncrementalKeyedStateHandle) keyedStateHandle)
                .map(IncrementalKeyedStateHandle::getMetaStateHandle)
                .map(this::getPath);
        return Stream.concat(rawStream, metadataStream).collect(Collectors.toSet());
    }
    private String getPath(StreamStateHandle streamStateHandle) {
        if (streamStateHandle instanceof FileStateHandle) {
            return ((FileStateHandle) streamStateHandle).getFilePath().toString();
        } else if (streamStateHandle instanceof ByteStreamStateHandle) {
            return ((ByteStreamStateHandle) streamStateHandle).getHandleName();
        }
        return null;
    }
}