Interactions Between Flink and HDFS

Understanding the interactions between Flink and HDFS helps clarify the problems HDFS may introduce into Flink.

Job Submission

When a client submits a Job, HDFS serves as an intermediate file-transfer location between the client and the AM. The files transferred are:

• flink jar: the conf and jar files in the Flink directory.
• ship files: files the user specifies for transfer to the cluster.
• flink-conf: submission adds new key-value pairs to the configuration, rewrites it to a file, and uploads it for the JM and TM to read.
• user jar: the jar containing the user’s code.
• jobGraph: only in per-job mode is the JobGraph serialized to disk; the JM then reads it from the file and submits it.

Blob Service

Blob Service stores large binary objects such as jars and log files. In practice, this means serializing them to HDFS. Blob Service is used in these scenarios:

• Jars required by jobs uploaded through REST are persisted to HDFS through Blob Service. Tasks then retrieve the jars to create their ClassLoaders.
• Logs displayed in the Web UI are first written to HDFS through Blob Service and then read back.

JobGraphStore

JobGraphStore supports HA: after a JobManager fails, another JobManager can retrieve the previous JobManager’s JobGraph from storage and resubmit it.

• Per-job mode: there is no JobGraphStore in this case (strictly speaking, one exists but does not persist data), because, as described above, per-job mode passes the JobGraph through a persisted jobGraph file.
• Session mode: with HA enabled, a JobGraphStore based on ZooKeeper and HDFS is used. In simple terms, it persists the graph to HDFS and stores its path (Handle) in ZooKeeper. A new JobManager can recover the JobGraph directly from that ZooKeeper path.

Checkpoint

The boundary between Flink’s Checkpoints and Savepoints has never been entirely clear, leading to some odd behavior. There are two cases:

Checkpointing enabled

This case clearly requires HDFS and needs no further explanation.

Checkpointing disabled

Even when checkpointing is disabled, Flink still initializes the CheckpointCoordinator and creates the corresponding checkpoint directory on HDFS. Intuitively, checkpoint-related components should not be initialized when checkpointing is off. However, they still need to be initialized to support manually triggering savepoints from the command line.