class SumItems: SerializableTransformer<KeyValue<String, Integer>, KeyValue<String, Integer>> {
    private val hm = HashMap<String, Integer>()

    override fun setup(context: Context) {

    }

    override fun cleanup() {

    }

    override fun transform(kv: KeyValue<String, Integer>, consumer: Consumer<KeyValue<String, Integer>>) {
        if (hm[kv.key] == null ) {
            hm[kv.key] = 0
        }
        hm[kv.key] += kv.getValue()
        consumer.accept(kv)
    }
}