Why Flink task heap is used periodically, even though no coming data?

Viewed 8

I found my Flink job is strange. The task heap was being used, but it didn't consume any data since it started.

It's my code, a stateful processing with three state:

  var listState: ListState[String] = _

  var valueState: ValueState[TestModel]  = _

  var mapState: MapState[String, TestModel] = _

  override def open(parameters: Configuration): Unit = {
    
    val descriptor = new ListStateDescriptor[String](
      "buffered-elements",
      createTypeInformation[String]
    )
    listState = getRuntimeContext.getListState(descriptor)

    val adStateDescriptor = new ValueStateDescriptor("LastEventState",  createTypeInformation[TestModel])
    valueState = getRuntimeContext.getState(adStateDescriptor)

    val mapStateDescriptor = new MapStateDescriptor("SubsegmentState",
      BasicTypeInfo.STRING_TYPE_INFO, createTypeInformation[TestModel])
    mapState = getRuntimeContext.getMapState(mapStateDescriptor)
  }

  override def map(in: TestModel): TestModel = {
    ......
  }

TaskManager memory is 6GB, and the Task heap size is 2.3GB

It's the graph of heap usage, the blue line is my task

0 Answers
Related