flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Michele Bertoni <michele1.bert...@mail.polimi.it>
Subject reduce error
Date Thu, 08 Oct 2015 08:53:12 GMT
Hi everybody, I am facing a very strange problem
Sometimes when I run my program one line of the result is totally wrong and if I repeat the
execution with same input it can change

The algorithm takes two dataset and execute something like a left outer join where
if there is a match
 - it increments a counter on the left tuple
 - add some values of the right tuple to an array of the left tuple
in the end it outputs the left value with the final counter and list of values (printer lines
A and B)

then there is a group reduce phase that merge element from different groups (there can be
but not always)

finally there is a map that applies some custom functions (printer lines C and D)

my problem is that from printer B and C the data actually change with no reason

in example in line B counter is 7 in line C it is 1.82E9, it always goes to very high numbers
from E9 to E18 while it is expected to be smaller than 20
moreover the array of extradata goes to null

what i found is that
it never happened if the reduce is actually executed (when there are many elements in a group)
but only when there is one element per group (so the reduce should not change data in a tuple)
I am running on a small dataset 60k lines and if it happens, it always happens in the same
line of the input
if I change dataset it happens on an other line but always on the same for that dataset

By now it always happened in only one line for each dataset (I am using a local environment
with parallelism 1)

I tried to changed the reduce with a reducegroup and inside the reducegroup a pure scala reduce,
this worked perfectly (by now)
I also added two groupby and reducegroup block (that take the line at put it in output) before
and after the groupby-reduce, in both cases i faced no problem at all
it only seems to appear when using the group and reduce

any idea on what could be the problem? I am getting mad on it

thanks for help

this is the code I use (there are a lot of println for debugging)

groupedRef // expID, bin, chromosome
  (references : Iterator[(Long, Long, Int, Int, Long, String, Long, Long, Char, Array[GValue],
Long)], experiments : Iterator[(Int, Int, Long, String, Long, Long, Char, Array[GValue])],
out : Collector[(Long, String, Long, Long, Char, Array[GValue], Array[List[GValue]], Int,
Long)]) => {
    val refCollected : List[PartialResult] = references.map((r) => (new PartialResult(r,
0, extraData ))).toList
    for(e <- experiments){
      for(r <- refCollected){
        if( /* complex clause not important */ ) {
          r.count += 1
          r.extra = r.extra :+ e._8.foldLeft(new Array[List[GValue]](0))((z : Array[List[GValue]],
v : GValue) => z :+ List(v))

    refCollected.map((pr) => {

      if(pr.binnedRegion._1.equals(7773719163500064339L) && pr.binnedRegion._7.equals(4860L)
&& pr.binnedRegion._8.equals(5324L)){
        println("-A--------------------------------------- " + pr.toString())
      val res = (pr.binnedRegion._1, pr.binnedRegion._6, pr.binnedRegion._7, pr.binnedRegion._8,
pr.binnedRegion._9, pr.binnedRegion._10, pr.extra.reduce((a,b) => a.zip(b).map((p) =>
p._1 ++ p._2)), pr.count, pr.binnedRegion._11)

      if(res._1.equals(7773719163500064339L) && res._3.equals(4860L) && res._4.equals(5324L)){
        println("-B--------------------------------------- " + (res._1, res._2, res._3, res._4,
res._5, res._6.mkString((" - ")), res._7.mkString(" - "), res._8, res._9 ))



val aggregationResult : DataSet[(Long, String, Long, Long, Char, Array[GValue])] =


//reduce phase
//concatenation of extra data
  (r1,r2) => {
    val out = (r1._1, r1._2, r1._3, r1._4, r1._5, r1._6,
        .map((a) => a._1 ++ a._2),
      r1._8 + r2._8, r1._9)




.map((l : (Long, String, Long, Long, Char, Array[GValue], Array[List[GValue]], Int, Long))
=> {

    if(l._1.equals(7773719163500064339L) && l._3.equals(4860L) && l._4.equals(5324L)){
      println("-C--------------------------------------- " + (l._1, l._2, l._3, l._4, l._5,
l._6.mkString((" - ")), l._7.mkString(" - "), l._8, l._9 ))

    val out =
      //something else that is working good

    if(out._1.equals(7773719163500064339L) && out._3.equals(4860L) && out._4.equals(5324L)){
      println("-D--------------------------------------- " + (out._1, out._2, out._3, out._4,
out._5, out._6.mkString(" - ")))


View raw message