Flink HDFS State Backend

[Jason Brelloch]

Hi everyone,

I am trying to set up flink with a hdfs state backend.  I configured state.backend and state.backend.fs.checkpointdir parameters in the flink-conf.yaml.  I run the flink task and the checkpoint directories are created in hdfs, so it appears it can connect and talk to hdfs just fine.  Unfortunately no files are ever created in the hdfs directory.  I checked that the state is being saved and restored from the task manager memory and that works fine, it just never writes to hdfs.

Am I missing a step?  Do I need to do anything to force a write to hdfs?  Does the state variable have to be a particular type to work with hdfs?

This is what my snapshot functions look like:

  override def restoreState (rState: scala.collection.mutable.HashMap[String, String]): Unit = {
    state = rState
  }

  override def snapshotState(checkpointId: Long, checkpointTimestamp: Long): scala.collection.mutable.HashMap[String, String] = {
    state
  }

Thanks!

-Jason

P.S.  I am running Flink v1.0.1, Hadoop 2.7.1, and Scala 2.11

[Aljoscha Krettek]

Hi,

could it be that your state is very small? In that case the state is not actually stored in HDFS but on the job manager because writing it to HDFS and storing a handle to that in the JobManager would be more expensive.

Cheers,

Aljoscha

On Mon, 18 Apr 2016 at 17:20 Jason Brelloch <[hidden email]> wrote:

Hi everyone,

I am trying to set up flink with a hdfs state backend.  I configured state.backend and state.backend.fs.checkpointdir parameters in the flink-conf.yaml.  I run the flink task and the checkpoint directories are created in hdfs, so it appears it can connect and talk to hdfs just fine.  Unfortunately no files are ever created in the hdfs directory.  I checked that the state is being saved and restored from the task manager memory and that works fine, it just never writes to hdfs.

Am I missing a step?  Do I need to do anything to force a write to hdfs?  Does the state variable have to be a particular type to work with hdfs?

This is what my snapshot functions look like:

  override def restoreState (rState: scala.collection.mutable.HashMap[String, String]): Unit = {
    state = rState
  }

  override def snapshotState(checkpointId: Long, checkpointTimestamp: Long): scala.collection.mutable.HashMap[String, String] = {
    state
  }

Thanks!

-Jason

P.S.  I am running Flink v1.0.1, Hadoop 2.7.1, and Scala 2.11

[Jason Brelloch]

Yep, that was the problem.

Thanks!

On Mon, Apr 18, 2016 at 11:36 AM, Aljoscha Krettek <[hidden email]> wrote:

Hi,

could it be that your state is very small? In that case the state is not actually stored in HDFS but on the job manager because writing it to HDFS and storing a handle to that in the JobManager would be more expensive.

Cheers,

Aljoscha

On Mon, 18 Apr 2016 at 17:20 Jason Brelloch <[hidden email]> wrote:

Hi everyone,

I am trying to set up flink with a hdfs state backend.  I configured state.backend and state.backend.fs.checkpointdir parameters in the flink-conf.yaml.  I run the flink task and the checkpoint directories are created in hdfs, so it appears it can connect and talk to hdfs just fine.  Unfortunately no files are ever created in the hdfs directory.  I checked that the state is being saved and restored from the task manager memory and that works fine, it just never writes to hdfs.

Am I missing a step?  Do I need to do anything to force a write to hdfs?  Does the state variable have to be a particular type to work with hdfs?

This is what my snapshot functions look like:

  override def restoreState (rState: scala.collection.mutable.HashMap[String, String]): Unit = {
    state = rState
  }

  override def snapshotState(checkpointId: Long, checkpointTimestamp: Long): scala.collection.mutable.HashMap[String, String] = {
    state
  }

Thanks!

-Jason

P.S.  I am running Flink v1.0.1, Hadoop 2.7.1, and Scala 2.11