import com.metamx.tranquility.typeclass.Timestamper; //导入依赖的package包/类
private Tranquilizer<Map<String, Object>> buildDruidService() {
    curatorFramework = buildCuratorFramework();
    final TimestampSpec timestampSpec = new TimestampSpec(timestampField, timestampFormat, null);
    final Timestamper<Map<String, Object>> timestamper = getTimestamper();
    if (timestamper != null) {
        final DruidLocation druidLocation = DruidLocation.create(indexService, firehosePattern, dataSource);
        final DruidRollup druidRollup = DruidRollup
                .create(DruidDimensions.specific(dimensions), aggregators, queryGranularity);
        final ClusteredBeamTuning clusteredBeamTuning = ClusteredBeamTuning.builder()
                .segmentGranularity(segmentGranularity)
                .windowPeriod(new Period(windowPeriod)).partitions(partitions).replicants(replicants).build();

        return DruidBeams.builder(timestamper).curator(curatorFramework).discoveryPath(discoveryPath).location(
                druidLocation).timestampSpec(timestampSpec).rollup(druidRollup).tuning(clusteredBeamTuning)
                .buildTranquilizer();
    } else {
        LOG.error("Building druid service error.");
        return null;
    }

}
 

import com.metamx.tranquility.typeclass.Timestamper; //导入依赖的package包/类
protected Timestamper<Map<String, Object>> getTimestamper() {
    return new Timestamper<Map<String, Object>>() {
        @Override
        public DateTime timestamp(Map<String, Object> theMap) {
            try {
                if (timestampFormat.equals("millis")) {
                    return new DateTime(Long.valueOf((String) theMap.get(timestampField)));
                } else {
                    return dateTimeFormatter.parseDateTime((String) theMap.get(timestampField));
                }
            } catch (Exception e) {
                LOG.error("Timestamp processing error.", e);
                // TODO: To verify behavior when datetime is null.
                return null;
            }
        }
    };
}
 

import com.metamx.tranquility.typeclass.Timestamper; //导入依赖的package包/类
@Test
public void dateFieldShouldBeProcessed() {
    final byte[] bytes = new byte[20];
    new Random().nextBytes(bytes);

    final Map<String, Object> headers = new HashMap<String, Object>();
    headers.put("date", Long.toString(System.currentTimeMillis()));
    headers.put("space", Arrays.toString(bytes));

    Timestamper<Map<String, Object>> timestamper = _tranquilitySink.getTimestamper();
    DateTime dateTime = timestamper.timestamp(headers);
    DateTime now = DateTime.now();
    Assert.assertTrue(dateTime.getYear() == now.getYear() && dateTime.getMonthOfYear() == now.getMonthOfYear()
            && dateTime.getDayOfMonth() == now.getDayOfMonth());

    headers.clear();
    headers.put("date", DateTime.now().toString());
    headers.put("space", Arrays.toString(bytes));

    Assert.assertTrue(dateTime.getYear() == now.getYear() && dateTime.getMonthOfYear() == now.getMonthOfYear()
            && dateTime.getDayOfMonth() == now.getDayOfMonth());
}
 

import com.metamx.tranquility.typeclass.Timestamper; //导入依赖的package包/类
@Test(expected = NullPointerException.class)
public void dateFieldShouldNotBeProcessed() {
    final byte[] bytes = new byte[20];
    new Random().nextBytes(bytes);

    final Map<String, Object> headers = new HashMap<String, Object>();
    headers.put("space", Arrays.toString(bytes));

    Timestamper<Map<String, Object>> timestamper = _tranquilitySink.getTimestamper();
    // Null Pointer Exception expected
    DateTime dateTime = timestamper.timestamp(headers);
    DateTime now = DateTime.now();
    Assert.assertTrue(dateTime.getYear() == now.getYear() && dateTime.getMonthOfYear() == now.getMonthOfYear()
            && dateTime.getDayOfMonth() == now.getDayOfMonth());

    dateTime = timestamper.timestamp(null);
    Assert.assertNull(dateTime);
}
 

import com.metamx.tranquility.typeclass.Timestamper; //导入依赖的package包/类
@Override
public Beam<Map<String, Object>> makeBeam(Map<?, ?> conf, IMetricsContext metrics) {

    List<AggregatorFactory> aggregator = getAggregatorList();

    // Tranquility needs to be able to extract timestamps from your object type (in this case, Map<String, Object>).
    final Timestamper<Map<String, Object>> timestamper = new StreamlineTimestamper(timestampField);

    // Tranquility uses ZooKeeper (through Curator) for coordination.
    final CuratorFramework curator = CuratorFrameworkFactory
            .builder()
            .connectString(tranquilityZKconnect) // we can use Storm conf to get config values
            .retryPolicy(new ExponentialBackoffRetry(1000, 20, 30000))
            .build();
    curator.start();

    // The JSON serialization of your object must have a timestamp field in a format that Druid understands. By default,
    // Druid expects the field to be called "timestamp" and to be an ISO8601 timestamp.
    final TimestampSpec timestampSpec = new TimestampSpec(timestampField, "auto", null);

    // Tranquility needs to be able to serialize your object type to JSON for transmission to Druid. By default this is
    // done with Jackson. If you want to provide an alternate serializer, you can provide your own via ```.objectWriter(...)```.
    // In this case, we won't provide one, so we're just using Jackson.
    final Beam<Map<String, Object>> beam = DruidBeams
            .builder(timestamper)
            .curator(curator)
            .discoveryPath(discoveryPath)
            .location(DruidLocation.create(indexService, dataSource))
            .timestampSpec(timestampSpec)
            .rollup(DruidRollup.create(DruidDimensions.specific(getTrimmedDimensions(dimensions)), aggregator, getQueryGranularity()))
            .tuning(
                    ClusteredBeamTuning
                            .builder()
                            .segmentGranularity(getSegmentGranularity())
                            .windowPeriod(new Period(windowPeriod))
                            .partitions(clusterPartitions)
                            .replicants(clusterReplication)
                            .build()
            )
            .druidBeamConfig(
                    DruidBeamConfig
                            .builder()
                            .indexRetryPeriod(new Period(indexRetryPeriod))
                            .build())
            .buildBeam();

    return beam;
}
 

import com.metamx.tranquility.typeclass.Timestamper; //导入依赖的package包/类
private Service buildDruidService() {
    curator = buildCurator();
    final TimestampSpec timestampSpec = new TimestampSpec(timestampField, "auto");
    final Timestamper<Map<String, Object>> timestamper = getTimestamper();
    final DruidLocation druidLocation = DruidLocation.create(indexService, firehosePattern, dataSource);
    final DruidRollup druidRollup = DruidRollup
            .create(DruidDimensions.specific(dimensions), aggregators, queryGranularity);
    final ClusteredBeamTuning clusteredBeamTuning = ClusteredBeamTuning.builder()
            .segmentGranularity(segmentGranularity)
            .windowPeriod(new Period(period)).partitions(partitions).replicants(replicants).build();//TODO revise

    return DruidBeams.builder(timestamper).curator(curator).discoveryPath(discoveryPath).location(
            druidLocation).timestampSpec(timestampSpec).rollup(druidRollup).tuning(clusteredBeamTuning)
            .buildJavaService();
}
 

import com.metamx.tranquility.typeclass.Timestamper; //导入依赖的package包/类
private Timestamper<Map<String, Object>> getTimestamper() {
    return new Timestamper<Map<String, Object>>() {
        @Override
        public DateTime timestamp(Map<String, Object> theMap) {
            return new DateTime(theMap.get(timestampField));
        }
    };
}