import io.druid.data.input.impl.TimestampSpec; //导入依赖的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 io.druid.data.input.impl.TimestampSpec; //导入依赖的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 io.druid.data.input.impl.TimestampSpec; //导入依赖的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();
}