package eu.dnetlib.dhp.contextpropagation

import org.apache.spark.sql.{Encoder, Encoders}
import org.apache.spark.sql.expressions.Aggregator

object PropagationAggregator {

  def getDatasetAggregator(): Aggregator[(String, PropagationStructure), PropagationStructure, PropagationStructure] = new Aggregator[(String, PropagationStructure), PropagationStructure, PropagationStructure]{

    override def zero: PropagationStructure = new PropagationStructure()

    override def reduce(b: PropagationStructure, a: (String, PropagationStructure)): PropagationStructure = {
      b.mergeFrom(a._2)


    }

    override def merge(wx: PropagationStructure, wy: PropagationStructure): PropagationStructure = {

      wx.mergeFrom(wy)

    }
    override def finish(reduction: PropagationStructure): PropagationStructure = reduction

    override def bufferEncoder: Encoder[PropagationStructure] =
      Encoders.kryo(classOf[PropagationStructure])

    override def outputEncoder: Encoder[PropagationStructure] =
      Encoders.kryo(classOf[PropagationStructure])



  }


}
