Beam<Map<String, Object>> buildBeam(String dataSource, String indexService, String discoveryPath, int clusterPartitions, int clusterReplication,
                                    String segmentGranularity, String queryGranularity, String windowPeriod, String firehoseGracePeriod, String indexRetryPeriod, DruidDimensions dimensions,
                                    List<AggregatorFactory> aggregator, Timestamper<Map<String, Object>> timestamper, TimestampSpec timestampSpec) {
    return DruidBeams.builder(timestamper)
            .curator(curator)
            .discoveryPath(discoveryPath)
            .location(DruidLocation.create(DruidEnvironment.create(indexService, FIREHOSE_PATTERN), dataSource))
            .timestampSpec(timestampSpec)
            .rollup(DruidRollup.create(dimensions, aggregator, QueryGranularity.fromString(queryGranularity)))
            .tuning(
                    ClusteredBeamTuning
                            .builder()
                            .segmentGranularity(getGranularity(segmentGranularity))
                            .windowPeriod(new Period(windowPeriod))
                            .partitions(clusterPartitions)
                            .replicants(clusterReplication)
                            .build()
            )
            .druidBeamConfig(
                    DruidBeamConfig
                            .builder()
                            .indexRetryPeriod(new Period(indexRetryPeriod))
                            .firehoseGracePeriod(new Period(firehoseGracePeriod))
                            .build())
            .buildBeam();
}
 

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();
}
 

@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;
}