import com.metamx.common.Granularity; //导入依赖的package包/类
public static Comparator<DataSegment> bucketMonthComparator()
{
  return new Comparator<DataSegment>()
  {
    @Override
    public int compare(DataSegment lhs, DataSegment rhs)
    {
      int retVal;

      DateTime lhsMonth = Granularity.MONTH.truncate(lhs.getInterval().getStart());
      DateTime rhsMonth = Granularity.MONTH.truncate(rhs.getInterval().getStart());

      retVal = lhsMonth.compareTo(rhsMonth);

      if (retVal != 0) {
        return retVal;
      }

      return lhs.compareTo(rhs);
    }
  };
}
 

import com.metamx.common.Granularity; //导入依赖的package包/类
@Override
public void configure(Context context) {

    indexService = context.getString(INDEX_SERVICE);
    discoveryPath = context.getString(DISCOVERY_PATH);
    dimensions = Arrays.asList(context.getString(DIMENSIONS).split(","));
    firehosePattern = context.getString(FIREHOSE_PATTERN, DEFAULT_FIREHOSE);
    dataSource = context.getString(DATA_SOURCE, DEFAUL_DATASOURCE);
    aggregators = AggregatorsHelper.build(context.getString(AGGREGATORS));
    queryGranularity = QueryGranularityHelper.getGranularity(context.getString(QUERY_GRANULARITY,
            DEFAULT_QUERY_GRANULARITY));
    segmentGranularity = Granularity.valueOf(context.getString(SEGMENT_GRANULARITY, DEFAULT_SEGMENT_GRANULARITY));
    period = context.getString(WINDOW_PERIOD, DEFAULT_PERIOD);
    partitions = context.getInteger(PARTITIONS, DEFAULT_PARTITIONS);
    replicants = context.getInteger(REPLICANTS, DEFAULT_REPLICANTS);
    // Tranquility needs to be able to extract timestamps from your object type (in this case, Map<String, Object>).
    timestampField = context.getString(TIMESTAMP_FIELD, DEFAULT_TIMESTAMP_FIELD);
    zookeeperLocation = context.getString(ZOOKEEPER_LOCATION, DEFAULT_ZOOKEEPER_LOCATION);
    baseSleppTime = context.getInteger(ZOOKEEPPER_BASE_SLEEP_TIME, DEFAULT_ZOOKEEPER_BASE_SLEEP);
    maxRetries = context.getInteger(ZOOKEEPER_MAX_RETRIES, DEFAULT_ZOOKEEPER_MAX_RETRIES);
    maxSleep = context.getInteger(ZOOKEEPER_MAX_SLEEP, DEFAULT_ZOOKEEPER_MAX_SLEEP);
    batchSize = context.getInteger(BATCH_SIZE, DEFAULT_BATCH_SIZE);

    druidService = buildDruidService();
    sinkCounter = new SinkCounter(this.getName());
    eventParser = new EventParser(timestampField);
}
 

import com.metamx.common.Granularity; //导入依赖的package包/类
@Override
public void configure(Context context) {
    indexService = context.getString(INDEX_SERVICE);
    discoveryPath = context.getString(DISCOVERY_PATH);
    dimensions = Arrays.asList(context.getString(DIMENSIONS).split(","));
    firehosePattern = context.getString(FIREHOSE_PATTERN, DEFAULT_FIREHOSE);
    dataSource = context.getString(DATA_SOURCE, DEFAUL_DATASOURCE);
    aggregators = AggregatorsHelper.build(context.getString(AGGREGATORS));
    queryGranularity = QueryGranularity.fromString(context.getString(QUERY_GRANULARITY, DEFAULT_QUERY_GRANULARITY));
    segmentGranularity = Granularity.valueOf(context.getString(SEGMENT_GRANULARITY, DEFAULT_SEGMENT_GRANULARITY));
    windowPeriod = context.getString(WINDOW_PERIOD, DEFAULT_PERIOD);
    partitions = context.getInteger(PARTITIONS, DEFAULT_PARTITIONS);
    replicants = context.getInteger(REPLICANTS, DEFAULT_REPLICANTS);
    timestampField = context.getString(TIMESTAMP_FIELD, DEFAULT_TIMESTAMP_FIELD);
    timestampFormat = context.getString(TIMESTAMP_FORMAT, null);
    zookeeperLocation = context.getString(ZOOKEEPER_LOCATION, DEFAULT_ZOOKEEPER_LOCATION);
    baseSleepTime = context.getInteger(ZOOKEEPPER_BASE_SLEEP_TIME, DEFAULT_ZOOKEEPER_BASE_SLEEP);
    maxRetries = context.getInteger(ZOOKEEPER_MAX_RETRIES, DEFAULT_ZOOKEEPER_MAX_RETRIES);
    maxSleep = context.getInteger(ZOOKEEPER_MAX_SLEEP, DEFAULT_ZOOKEEPER_MAX_SLEEP);
    batchSize = context.getInteger(BATCH_SIZE, DEFAULT_BATCH_SIZE);

    druidService = sinkStrategy.getDruidService();
    if (druidService != null) {
        sinkCounter = new SinkCounter(this.getName());
        if (timestampFormat.equals("auto"))
            dateTimeFormatter = DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss'Z'");
        else if (timestampFormat.equals("millis"))
            dateTimeFormatter = null;
        else
            dateTimeFormatter = DateTimeFormat.forPattern(timestampFormat);

        // Filter defined fields
        Set<String> filter = new HashSet<String>();
        filter.add(timestampField);
        filter.addAll(dimensions);
        for (AggregatorFactory aggregatorFactory : aggregators) {
            filter.addAll(aggregatorFactory.requiredFields());
        }

        eventParser = new FlumeEventParser(timestampField, dateTimeFormatter, filter);
    }
}
 

import com.metamx.common.Granularity; //导入依赖的package包/类
private Granularity getSegmentGranularity() {
    Granularity granularity = Granularity.HOUR;

    switch (segmentGranularity) {
        case "SECOND":
            granularity = Granularity.SECOND;
            break;
        case "MINUTE":
            granularity = Granularity.MINUTE;
            break;
        case "FIVE_MINUTE":
            granularity = Granularity.FIVE_MINUTE;
            break;
        case "TEN_MINUTE":
            granularity = Granularity.TEN_MINUTE;
            break;
        case "FIFTEEN_MINUTE":
            granularity = Granularity.FIFTEEN_MINUTE;
            break;
        case "HOUR":
            granularity = Granularity.HOUR;
            break;
        case "SIX_HOUR":
            granularity = Granularity.SIX_HOUR;
            break;
        case "DAY":
            granularity = Granularity.DAY;
            break;
        case "WEEK":
            granularity = Granularity.WEEK;
            break;
        case "MONTH":
            granularity = Granularity.MONTH;
            break;
        case "YEAR":
            granularity = Granularity.YEAR;
            break;
        default:
            break;
    }
    return granularity;
}