@Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
        this.outputCollector = basicOutputCollector;
        String imsi = tuple.getString(0);
        String eventType = tuple.getString(1);
        Long time = tuple.getLong(2);
        String lac = tuple.getString(3);
        String cell = tuple.getString(4);
        logger.info(String.format("[%s]%s,%s,%s,%s,%s", PRECONDITION, imsi, eventType, time, lac, cell));
        if (logger.isDebugEnabled()) {
            logger.debug(String.format("[%s]%s,%s,%s,%s,%s", PRECONDITION, imsi, eventType, time, lac, cell));
        }

        boolean matchARPU = KbUtils.getInstance().checkARPU(imsi);
        if (matchARPU) {
            outputCollector.emit(PRECONDITION, new Values(imsi, eventType, time, lac, cell));
        } else {
            if (logger.isDebugEnabled()) {
                logger.debug(String.format("ARPU not match:%s", imsi));
            }
        }

        if ((i & 1023) == 0) { // 隔1024条
//            if (time > lastSignalTime){
            outputCollector.emit(UPDATETIME, new Values(time, imsi));
//                lastSignalTime = time;
            if (logger.isInfoEnabled()) {
                logger.info(String.format("[%s]%s/%s", UPDATETIME, time, TimeUtil.getTime(time)));
            }
//            }
        }
        i++;

//        if (time > lastSignalTime){
//            basicOutputCollector.emit(UPDATETIME, new Values(time, imsi));
//            lastSignalTime = time;
//            logger.debug(String.format("[%s]%s", UPDATETIME, time));
//        }
    }