Updated on 2024-08-16 GMT+08:00

Scala Sample Code of Flink Asynchronous Checkpoint

Sample Code

Assume that you want to collect data volume in a 4-second time window every other second and the status of operators must be strictly consistent.

  • Formats of sent data
    1
    case class SEvent(id: Long, name: String, info: String, count: Int)
    
  • Snapshot data

    The snapshot data is used to store number of data pieces recorded by operators during creation of snapshots.

     1
     2
     3
     4
     5
     6
     7
     8
     9
    10
        // User-defined status
    class UDFStateScala extends Serializable{
      private var count = 0L
    
      // Set the user-defined status.
      def setState(s: Long) = count = s
    
      // Obtain the user-defined status.
      def getState = count
    }
    
  • Data source with checkpoints

    The code snippet of a source operator pauses 1 second every time after sending 10,000 pieces of data. When a snapshot is created, the code saves the total number of sent data pieces in UDFState. When the snapshot is used for restoration, the number of sent data pieces saved in UDFState is read and assigned to the count variable.

     1
     2
     3
     4
     5
     6
     7
     8
     9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    import java.util
    import org.apache.flink.streaming.api.checkpoint.ListCheckpointed
    import org.apache.flink.streaming.api.functions.source.RichSourceFunction
    import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
    
    case class SEvent(id: Long, name: String, info: String, count: Int)
    
    // This class is a source operator with a checkpoint.
    class SEventSourceWithChk extends RichSourceFunction[SEvent] with ListCheckpointed[UDFStateScala]{
      private var count = 0L
      private var isRunning = true
      private val alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWZYX0987654321"
    
      // The logic of a source operator is to inject 10,000 tuples to the StreamGraph every second.
      override def run(sourceContext: SourceContext[SEvent]): Unit = {
        while(isRunning) {
          for (i <- 0 until 10000) {
            sourceContext.collect(SEvent(1, "hello-"+count, alphabet,1))
            count += 1L
          }
          Thread.sleep(1000)
        }
      }
    
      // Invoked when a task is canceled
      override def cancel(): Unit = {
        isRunning = false;
      }
    
      override def close(): Unit = super.close()
    
      // Create a snapshot.
      override def snapshotState(l: Long, l1: Long): util.List[UDFStateScala] = {
        val udfList: util.ArrayList[UDFStateScala] = new util.ArrayList[UDFStateScala]
        val udfState = new UDFStateScala
        udfState.setState(count)
        udfList.add(udfState)
        udfList
      }
    
      // Obtain the status from the snapshot.
      override def restoreState(list: util.List[UDFStateScala]): Unit = {
        val udfState = list.get(0)
        count = udfState.getState
      }
    }
    
  • Definition of a window with a checkpoint

    This code snippet is about a window operator and is used to calculate the number or tuples in a window.

     1
     2
     3
     4
     5
     6
     7
     8
     9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    import java.util
    import org.apache.flink.api.java.tuple.Tuple
    import org.apache.flink.streaming.api.checkpoint.ListCheckpointed
    import org.apache.flink.streaming.api.scala.function.WindowFunction
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    // This class is a window operator with a checkpoint.
    class WindowStatisticWithChk extends WindowFunction[SEvent, Long, Tuple, TimeWindow] with ListCheckpointed[UDFStateScala]{
      private var total = 0L
    
      // Define the window operator implementation logic to calculate the number of tuples in a window.
      override def apply(key: Tuple, window: TimeWindow, input: Iterable[SEvent], out: Collector[Long]): Unit = {
        var count = 0L
        for (event <- input) {
          count += 1L
        }
        total += count
        out.collect(count)
      }
    
      // Create a snapshot for the user-defined status.
      override def snapshotState(l: Long, l1: Long): util.List[UDFStateScala] = {
        val udfList: util.ArrayList[UDFStateScala] = new util.ArrayList[UDFStateScala]
        val udfState = new UDFStateScala
        udfState.setState(total)
        udfList.add(udfState)
        udfList
      }
    
      // Restore the status from the user-defined snapshot.
      override def restoreState(list: util.List[UDFStateScala]): Unit = {
        val udfState = list.get(0)
        total = udfState.getState
      }
    }
    
  • Application code

    This code snippet is about the definition of StreamGraph and detailed service implementation process. The event time is used as time to trigger the window.

     1
     2
     3
     4
     5
     6
     7
     8
     9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    import org.apache.flink.runtime.state.filesystem.FsStateBackend
    import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
    import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.api.watermark.Watermark
    import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    
    object FlinkEventTimeAPIChkMain {
      def main(args: Array[String]): Unit ={
        val chkPath = ParameterTool.fromArgs(args).get("chkPath", "hdfs://hacluster/flink/checkpoint/checkpoint/")
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setStateBackend(new FsStateBackend(chkPath))
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        env.getConfig.setAutoWatermarkInterval(2000)
        env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
        env.getCheckpointConfig.setCheckpointInterval(6000)
    
        // Application logic
        env.addSource(new SEventSourceWithChk)
          .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[SEvent] {
            // Set a watermark.
            override def getCurrentWatermark: Watermark = {
              new Watermark(System.currentTimeMillis())
            }
           // Add a timestamp to each tuple.
            override def extractTimestamp(t: SEvent, l: Long): Long = {
              System.currentTimeMillis()
            }
          })
          .keyBy(0)
          .window(SlidingEventTimeWindows.of(Time.seconds(4), Time.seconds(1)))
          .apply(new WindowStatisticWithChk)
          .print()
        env.execute()
      }
    }