package com.kuaiyou.asmgo.hadoop.searchterm;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import com.kuaiyou.asmgo.hadoop.searchterm.model.Searchterm;
import com.kuaiyou.asmgo.util.JacksonUtil;
import com.kuaiyou.asmgo.util.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class SearchtermReducer extends Reducer<Text, Text, NullWritable, Text> {

    private Text outText = new Text();
    private Logger logger = LoggerFactory.getLogger(getClass());

    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Iterator<Text> iterator = values.iterator();
        Text last = null;
        if (iterator != null) {
            while (iterator.hasNext()) {
                // 只保留最后一个，会存在同一个词N条记录的情况
                last = iterator.next();
            }
        }
        try {
            if (last != null) {
                String[] reduKey = key.toString().split(StringUtil.SPLIT_UNDERLINE);
                Searchterm entity = new Searchterm();
                entity.setTerm(reduKey[0]);
                entity.setLang(reduKey[1]);
                entity.setPriority(Integer.valueOf(reduKey[2]));
                String appStr = last.toString();

                // |app1|app2 转成list
                List<String> apps = new ArrayList<>();
                String[] splits = appStr.split(StringUtil.SPLIT_VERTICALLINE_);
                for (String item : splits) {
                    if (StringUtils.isNotEmpty(item)) {
                        apps.add(item);
                    }
                }
                entity.setApps(apps);
                // 序列化到json字符串
                String outvalue = JacksonUtil.writeValueAsString(entity);
                outText.set(outvalue);
                context.write(NullWritable.get(), outText);

            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
}