Nguồn dữ liệu: Lưu trữ ExecutionGraphInfo từ Flink Runtime

Khi job dừng hoặc gặp lỗi, HistoryServerArchivist được gọi để lưu trữ:

public interface HistoryServerArchivist {
    CompletableFuture<Acknowledge> archiveExecutionGraph(ExecutionGraphInfo executionGraphInfo);
}

Điểm gọi chính

protected CompletableFuture<CleanupJobState> jobReachedTerminalState(
        ExecutionGraphInfo executionGraphInfo) {
}

private CompletableFuture<Acknowledge> archiveExecutionGraphToHistoryServer(
        ExecutionGraphInfo executionGraphInfo) {
}

Tuần tự hóa dữ liệu TaskManager

Ví dụ về cách dữ liệu TaskManager được tổ chức:

public class JobVertexTaskManagersHandler
        extends AbstractAccessExecutionGraphHandler<
                JobVertexTaskManagersInfo, JobVertexMessageParameters>
        implements OnlyExecutionGraphJsonArchivist {

    @Override
    public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)
            throws IOException {
        Collection<? extends AccessExecutionJobVertex> vertices = graph.getAllVertices().values();
        List<ArchivedJson> archive = new ArrayList<>(vertices.size());
        for (AccessExecutionJobVertex task : vertices) {
            ResponseBody json = createJobVertexTaskManagersInfo(task, graph.getJobID(), null);
            String path = getMessageHeaders()
                    .getTargetRestEndpointURL()
                    .replace(':' + JobIDPathParameter.KEY, graph.getJobID().toString())
                    .replace(':' + JobVertexIdPathParameter.KEY,
                            task.getJobVertexId().toString());
            archive.add(new ArchivedJson(path, json));
        }
        return archive;
    }
}

Cấu trúc dữ liệu cho Jackson

Cấu trúc dữ liệu tương ứng với JSON thực tế:

@JsonCreator
public TaskManagersInfo(
        @JsonProperty("host") String host,
        @JsonProperty("status") ExecutionState status,
        @JsonProperty("start-time") long startTime,
        @JsonProperty("end-time") long endTime,
        @JsonProperty("duration") long duration,
        @JsonProperty("metrics") IOMetricsInfo metrics,
        @JsonProperty("status-counts") Map<ExecutionState, Integer> statusCounts,
        @JsonProperty("taskmanager-id") String taskmanagerId,
        @JsonProperty("aggregated") AggregatedTaskDetailsInfo aggregated) {
    // Khởi tạo các trường
}

FsJobArchivist tuần tự hóa dữ liệu thành JSON

Dữ liệu được ghi vào thư mục được chỉ định bởi jobmanager.archive.fs.dir:

public static Path archiveJob(
        Path rootPath, JobID jobId, Collection<ArchivedJson> jsonToArchive) throws IOException {
    FileSystem fs = rootPath.getFileSystem();
    Path path = new Path(rootPath, jobId.toString());
    OutputStream out = fs.create(path, FileSystem.WriteMode.NO_OVERWRITE);
    
    try (JsonGenerator gen = jacksonFactory.createGenerator(out, JsonEncoding.UTF8)) {
        gen.writeStartObject();
        gen.writeArrayFieldStart("archive");
        for (ArchivedJson archive : jsonToArchive) {
            gen.writeStartObject();
            gen.writeStringField("path", archive.getPath());
            gen.writeStringField("json", archive.getJson());
            gen.writeEndObject();
        }
        gen.writeEndArray();
        gen.writeEndObject();
    }
    return path;
}

Cấu trúc dữ liệu JSON lưu trữ

Dữ liệu lưu trữ có cấu trúc JSON như sau:

{
    "archive": [
        {
            "path": "/jobs/overview",
            "json": "..."
        },
        {
            "path": "/jobs/{jobid}/config",
            "json": "..."
        },
        {
            "path": "/jobs/{jobid}/checkpoints",
            "json": "..."
        }
    ]
}

HistoryServer quét dữ liệu lưu trữ khi khởi động

public class HistoryServer {
    void start() throws IOException, InterruptedException {
        executor.scheduleWithFixedDelay(
                getArchiveFetchingRunnable(), 0, refreshIntervalMillis, TimeUnit.MILLISECONDS);
    }
    
    private Runnable getArchiveFetchingRunnable() {
        return () -> archiveFetcher.fetchArchives();
    }
}

Xử lý dữ liệu lưu trữ: Tải về đĩa cục bộ

private void processArchive(String jobID, Path jobArchive) throws IOException {
    for (ArchivedJson archive : FsJobArchivist.getArchivedJsons(jobArchive)) {
        String path = archive.getPath();
        String json = archive.getJson();
        
        File target;
        if (path.equals(JobsOverviewHeaders.URL)) {
            target = new File(webOverviewDir, jobID + ".json");
        } else {
            target = new File(webDir, path + ".json");
        }
        // Ghi dữ liệu vào file
    }
}

API trả về file tương ứng

router.addGet("/:*", new HistoryServerStaticFileServerHandler(webDir));

Các yêu cầu được chia thành hai loại:

• File JSON lưu trữ job
• Liên kết log của JobManager và TaskManager

Thông tin log Container

Để lấy thông tin TaskManager:

• Tìm path=/jobs/<jobid>/vertices/ để lấy tất cả vertexid
• Tìm path=/jobs/<jobid>/vertices/<vertexid>/taskmanagers để thiết lập ánh xạ

Cắt ngắn hostName TaskManager

HostNameSupplier chỉ trả về segment đầu tiên:

public class DefaultHostNameSupplier {
    public String getHostName() {
        // Chỉ trả về segment đầu tiên của FQDN
    }
}