Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-4586

NumberSequenceIterator and Accumulator threading issue

    Details

    • Type: Bug
    • Status: Closed
    • Priority: Minor
    • Resolution: Fixed
    • Affects Version/s: 1.1.2
    • Fix Version/s: 1.2.0, 1.1.4
    • Component/s: DataSet API
    • Labels:
      None

      Description

      There is a strange problem when using the NumberSequenceIterator in combination with an AverageAccumulator.

      It seems like the individual accumulators are reinitialized and overwrite parts of intermediate solutions.

      The following scala snippit exemplifies the problem.

      Instead of printing the correct average, the result should be 50.5 but is something completely different, like 8.08, dependent on the number of cores used.

      If the parallelism is set to 1 the result is correct, which indicates a likely threading problem.

      The problem occurs using the java and scala API.

      env
        .fromParallelCollection(new NumberSequenceIterator(1, 100))
        .map(new RichMapFunction[Long, Long] {
      	var a : AverageAccumulator = _
      
      	override def map(value: Long): Long = {
      	  a.add(value)
      	  value
      	}
      
      	override def open(parameters: Configuration): Unit = {
      	  a = new AverageAccumulator
      	  getRuntimeContext.addAccumulator("test", a)
      	}
        })
        .reduce((a, b) => a + b)
        .print()
      
      
      val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult
      
      println(lastJobExecutionResult.getAccumulatorResult("test"))
      

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                greghogan Greg Hogan
                Reporter:
                jkirsch Johannes
              • Votes:
                0 Vote for this issue
                Watchers:
                4 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: