<commit_before>## Tranquility

Tranquility helps you send event streams to Druid, the raddest data store ever (http://druid.io/), in real-time. It
handles partitioning, replication, service discovery, and schema rollover for you, seamlessly and without downtime.
Tranquility is written in Scala, and bundles idiomatic Java and Scala APIs that work nicely with Finagle, Storm, and
Trident.

This project is a friend of Druid. For discussion, feel free to use the normal Druid channels: http://druid.io/community.html

## Direct API

If you want to write a program that sends data to Druid, you'll likely end up using the direct Finagle-based API. (The
other alternative is the Storm API, described in the next section.)

You can set up and use a Finagle Service like this:

```java
final String indexService = "druid:overlord"; // Your overlord's service name.
final String firehosePattern = "druid:firehose:%s"; // Make up a service pattern, include %s somewhere in it.
final String discoveryPath = "/discovery"; // Your overlord's druid.discovery.curator.path
final String dataSource = "foo";
final List<String> dimensions = ImmutableList.of("bar", "qux");
final List<AggregatorFactory> aggregators = ImmutableList.of(
    new CountAggregatorFactory("cnt"),
    new LongSumAggregatorFactory("baz", "baz")
);

// 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 Timestamper<Map<String, Object>>()
{
    @Override
    public DateTime timestamp(Map<String, Object> theMap)
    {
        return new DateTime(theMap.get("timestamp"));
    }
};

// Tranquility uses ZooKeeper (through Curator) for coordination.
final CuratorFramework curator = CuratorFrameworkFactory
    .builder()
    .connectString("zk.example.com:2181")
    .retryPolicy(new ExponentialBackoffRetry(1000, 20, 30000))
    .build();
curator.start();

// Tranquility needs to be able to serialize your object type. By default this is done with Jackson. If you want to
// provide an alternate serializer, you can provide your own via ```.eventWriter(...)```. In this case, we won't
// provide one, so we're just using Jackson:
final Service<List<Map<String, Object>>, Integer> druidService = DruidBeams
    .builder(timestamper)
    .curator(curator)
    .discoveryPath(discoveryPath)
    .location(
        DruidLocation.create(
            indexService,
            firehosePattern,
            dataSource
        )
    )
    .rollup(DruidRollup.create(DruidDimensions.specific(dimensions), aggregators, QueryGranularity.MINUTE))
    .tuning(ClusteredBeamTuning.create(Granularity.HOUR, new Period("PT0M"), new Period("PT10M"), 1, 1))
    .buildJavaService();

// Send events to Druid:
final Future<Integer> numSentFuture = druidService.apply(listOfEvents);

// Wait for confirmation:
final Integer numSent = Await.result(numSentFuture);

// Close lifecycled objects:
Await.result(druidService.close());
curator.close();
```

Or in Scala:

```scala
val indexService = "druid:overlord" // Your overlord's druid.service, with slashes replaced by colons.
val firehosePattern = "druid:firehose:%s" // Make up a service pattern, include %s somewhere in it.
val discoveryPath = "/discovery" // Your overlord's druid.discovery.curator.path.
val dataSource = "foo"
val dimensions = Seq("bar", "qux")
val aggregators = Seq(new CountAggregatorFactory("cnt"), new LongSumAggregatorFactory("baz", "baz"))

// Tranquility needs to be able to extract timestamps from your object type (in this case, Map<String, Object>).
val timestamper = (eventMap: Map[String, Any]) => new DateTime(eventMap("timestamp"))

// Tranquility needs to be able to serialize your object type. By default this is done with Jackson. If you want to
// provide an alternate serializer, you can provide your own via ```.eventWriter(...)```. In this case, we won't
// provide one, so we're just using Jackson:
val druidService = DruidBeams
  .builder(timestamper)
  .curator(curator)
  .discoveryPath(discoveryPath)
  .location(DruidLocation(indexService, firehosePattern, dataSource))
  .rollup(DruidRollup(SpecificDruidDimensions(dimensions), aggregators, QueryGranularity.MINUTE))
  .tuning(ClusteredBeamTuning(Granularity.HOUR, 10.minutes, 1, 1))
  .buildService()

// Send events to Druid:
val numSentFuture: Future[Int] = druidService(listOfEvents)

// Wait for confirmation:
val numSent = Await.result(numSentFuture)
```

## Storm

If you're using Storm to generate your event stream, you can use Tranquility's builtin Bolt adapter. This Bolt expects
to receive tuples in which the zeroth element is your event type (in this case, Scala Maps). It does not emit any
tuples of its own.

It must be supplied with a BeamFactory. You can implement one of these using the DruidBeams builder's "buildBeam()"
method. For example:

```scala
class MyBeamFactory extends BeamFactory[Map[String, Any]]
{
  def makeBeam(conf: java.util.Map[_, _], metrics: IMetricsContext) = {
    // This means you'll need a "tranquility.zk.connect" property in your Storm topology.
    val curator = CuratorFrameworkFactory.newClient(
      conf.get("tranquility.zk.connect").asInstanceOf[String],
      new BoundedExponentialBackoffRetry(100, 1000, 5)
    )
    curator.start()

    val indexService = "druid:overlord" // Your overlord's druid.service, with slashes replaced by colons.
    val firehosePattern = "druid:firehose:%s" // Make up a service pattern, include %s somewhere in it.
    val discoveryPath = "/discovery" // Your overlord's druid.discovery.curator.path.
    val dataSource = "foo"
    val dimensions = Seq("bar")
    val aggregators = Seq(new LongSumAggregatorFactory("baz", "baz"))

    DruidBeams
      .builder((eventMap: Map[String, Any]) => new DateTime(eventMap("timestamp")))
      .curator(curator)
      .discoveryPath(discoveryPath)
      .location(DruidLocation(indexService, firehosePattern, dataSource))
      .rollup(DruidRollup(SpecificDruidDimensions(dimensions), aggregators, QueryGranularity.MINUTE))
      .tuning(ClusteredBeamTuning(Granularity.HOUR, 0.minutes, 10.minutes, 1, 1))
      .buildBeam()
  }
}

// Add this bolt to your topology:
val bolt = new BeamBolt(new MyBeamFactory)
```

## Trident

If you're using Trident on top of Storm, you can use Trident's partitionPersist in concert with Tranquility's
TridentBeamStateFactory (which takes a BeamFactory, like the Storm Bolt) and TridentBeamStateUpdater.

## JARs

Tranquility artifacts are hosted on the Metamarkets maven repository: https://metamx.artifactoryonline.com/metamx/pub-libs-releases-local/.
If you set up your project to know about this repository, you can depend on one of the hosted versions.

The current stable version is:

```xml
<dependency>
  <groupId>com.metamx</groupId>
  <artifactId>tranquility_2.10</artifactId>
  <!-- Or for scala 2.9: -->
  <!-- <artifactId>tranquility_2.9.1</artifactId> -->
  <version>0.2.1</version>
</dependency>
```

Tranquility is built with [SBT](http://www.scala-sbt.org/). If you want to build the jars yourself, you can
run ```sbt +package```. Tranquility can be cross-built for multiple Scala versions, so you will get one jar
for each Scala version we currently support.

## Storm Setup

Tranquility depends on a newer version of zookeeper than Storm is built with, at least through Storm 0.9.1. This should
be worked out once [STORM-70](https://issues.apache.org/jira/browse/STORM-70) is in a release, but for the time being,
Tranquility deployments will work better on a patched Storm. Our fork is here: https://github.com/metamx/incubator-storm/tree/v0.9.1-incubating-mmx.
We have also published artifacts in the the metamx maven repository at: https://metamx.artifactoryonline.com/metamx/pub-libs-releases-local/org/apache/storm/storm-core/.

## Druid Setup

Tranquility works with the Druid indexing service (http://druid.io/docs/latest/Indexing-Service.html). To get started,
you'll need an Overlord, enough Middle Managers for your realtime workload, and enough Historical nodes to receive
handoffs. You don't need any Realtime nodes, since Tranquility uses the indexing service for all of its ingestion needs.

Tranquility periodically submits new tasks to the indexing service to provide for log rotation and to support
zero-downtime configuration changes. These new tasks are typically submitted before the old ones exit, so to allow for<commit_msg>Update recommended version.<commit_after>## Tranquility

Tranquility helps you send event streams to Druid, the raddest data store ever (http://druid.io/), in real-time. It
handles partitioning, replication, service discovery, and schema rollover for you, seamlessly and without downtime.
Tranquility is written in Scala, and bundles idiomatic Java and Scala APIs that work nicely with Finagle, Storm, and
Trident.

This project is a friend of Druid. For discussion, feel free to use the normal Druid channels: http://druid.io/community.html

## Direct API

If you want to write a program that sends data to Druid, you'll likely end up using the direct Finagle-based API. (The
other alternative is the Storm API, described in the next section.)

You can set up and use a Finagle Service like this:

```java
final String indexService = "druid:overlord"; // Your overlord's service name.
final String firehosePattern = "druid:firehose:%s"; // Make up a service pattern, include %s somewhere in it.
final String discoveryPath = "/discovery"; // Your overlord's druid.discovery.curator.path
final String dataSource = "foo";
final List<String> dimensions = ImmutableList.of("bar", "qux");
final List<AggregatorFactory> aggregators = ImmutableList.of(
    new CountAggregatorFactory("cnt"),
    new LongSumAggregatorFactory("baz", "baz")
);

// 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 Timestamper<Map<String, Object>>()
{
    @Override
    public DateTime timestamp(Map<String, Object> theMap)
    {
        return new DateTime(theMap.get("timestamp"));
    }
};

// Tranquility uses ZooKeeper (through Curator) for coordination.
final CuratorFramework curator = CuratorFrameworkFactory
    .builder()
    .connectString("zk.example.com:2181")
    .retryPolicy(new ExponentialBackoffRetry(1000, 20, 30000))
    .build();
curator.start();

// Tranquility needs to be able to serialize your object type. By default this is done with Jackson. If you want to
// provide an alternate serializer, you can provide your own via ```.eventWriter(...)```. In this case, we won't
// provide one, so we're just using Jackson:
final Service<List<Map<String, Object>>, Integer> druidService = DruidBeams
    .builder(timestamper)
    .curator(curator)
    .discoveryPath(discoveryPath)
    .location(
        DruidLocation.create(
            indexService,
            firehosePattern,
            dataSource
        )
    )
    .rollup(DruidRollup.create(DruidDimensions.specific(dimensions), aggregators, QueryGranularity.MINUTE))
    .tuning(ClusteredBeamTuning.create(Granularity.HOUR, new Period("PT0M"), new Period("PT10M"), 1, 1))
    .buildJavaService();

// Send events to Druid:
final Future<Integer> numSentFuture = druidService.apply(listOfEvents);

// Wait for confirmation:
final Integer numSent = Await.result(numSentFuture);

// Close lifecycled objects:
Await.result(druidService.close());
curator.close();
```

Or in Scala:

```scala
val indexService = "druid:overlord" // Your overlord's druid.service, with slashes replaced by colons.
val firehosePattern = "druid:firehose:%s" // Make up a service pattern, include %s somewhere in it.
val discoveryPath = "/discovery" // Your overlord's druid.discovery.curator.path.
val dataSource = "foo"
val dimensions = Seq("bar", "qux")
val aggregators = Seq(new CountAggregatorFactory("cnt"), new LongSumAggregatorFactory("baz", "baz"))

// Tranquility needs to be able to extract timestamps from your object type (in this case, Map<String, Object>).
val timestamper = (eventMap: Map[String, Any]) => new DateTime(eventMap("timestamp"))

// Tranquility needs to be able to serialize your object type. By default this is done with Jackson. If you want to
// provide an alternate serializer, you can provide your own via ```.eventWriter(...)```. In this case, we won't
// provide one, so we're just using Jackson:
val druidService = DruidBeams
  .builder(timestamper)
  .curator(curator)
  .discoveryPath(discoveryPath)
  .location(DruidLocation(indexService, firehosePattern, dataSource))
  .rollup(DruidRollup(SpecificDruidDimensions(dimensions), aggregators, QueryGranularity.MINUTE))
  .tuning(ClusteredBeamTuning(Granularity.HOUR, 10.minutes, 1, 1))
  .buildService()

// Send events to Druid:
val numSentFuture: Future[Int] = druidService(listOfEvents)

// Wait for confirmation:
val numSent = Await.result(numSentFuture)
```

## Storm

If you're using Storm to generate your event stream, you can use Tranquility's builtin Bolt adapter. This Bolt expects
to receive tuples in which the zeroth element is your event type (in this case, Scala Maps). It does not emit any
tuples of its own.

It must be supplied with a BeamFactory. You can implement one of these using the DruidBeams builder's "buildBeam()"
method. For example:

```scala
class MyBeamFactory extends BeamFactory[Map[String, Any]]
{
  def makeBeam(conf: java.util.Map[_, _], metrics: IMetricsContext) = {
    // This means you'll need a "tranquility.zk.connect" property in your Storm topology.
    val curator = CuratorFrameworkFactory.newClient(
      conf.get("tranquility.zk.connect").asInstanceOf[String],
      new BoundedExponentialBackoffRetry(100, 1000, 5)
    )
    curator.start()

    val indexService = "druid:overlord" // Your overlord's druid.service, with slashes replaced by colons.
    val firehosePattern = "druid:firehose:%s" // Make up a service pattern, include %s somewhere in it.
    val discoveryPath = "/discovery" // Your overlord's druid.discovery.curator.path.
    val dataSource = "foo"
    val dimensions = Seq("bar")
    val aggregators = Seq(new LongSumAggregatorFactory("baz", "baz"))

    DruidBeams
      .builder((eventMap: Map[String, Any]) => new DateTime(eventMap("timestamp")))
      .curator(curator)
      .discoveryPath(discoveryPath)
      .location(DruidLocation(indexService, firehosePattern, dataSource))
      .rollup(DruidRollup(SpecificDruidDimensions(dimensions), aggregators, QueryGranularity.MINUTE))
      .tuning(ClusteredBeamTuning(Granularity.HOUR, 0.minutes, 10.minutes, 1, 1))
      .buildBeam()
  }
}

// Add this bolt to your topology:
val bolt = new BeamBolt(new MyBeamFactory)
```

## Trident

If you're using Trident on top of Storm, you can use Trident's partitionPersist in concert with Tranquility's
TridentBeamStateFactory (which takes a BeamFactory, like the Storm Bolt) and TridentBeamStateUpdater.

## JARs

Tranquility artifacts are hosted on the Metamarkets maven repository: https://metamx.artifactoryonline.com/metamx/pub-libs-releases-local/.
If you set up your project to know about this repository, you can depend on one of the hosted versions.

The current stable version is:

```xml
<dependency>
  <groupId>com.metamx</groupId>
  <artifactId>tranquility_2.10</artifactId>
  <!-- Or for scala 2.9: -->
  <!-- <artifactId>tranquility_2.9.1</artifactId> -->
  <version>0.2.12</version>
</dependency>
```

Tranquility is built with [SBT](http://www.scala-sbt.org/). If you want to build the jars yourself, you can
run ```sbt +package```. Tranquility can be cross-built for multiple Scala versions, so you will get one jar
for each Scala version we currently support.

## Storm Setup

Tranquility depends on a newer version of zookeeper than Storm is built with, at least through Storm 0.9.1. This should
be worked out once [STORM-70](https://issues.apache.org/jira/browse/STORM-70) is in a release, but for the time being,
Tranquility deployments will work better on a patched Storm. Our fork is here: https://github.com/metamx/incubator-storm/tree/v0.9.1-incubating-mmx.
We have also published artifacts in the the metamx maven repository at: https://metamx.artifactoryonline.com/metamx/pub-libs-releases-local/org/apache/storm/storm-core/.

## Druid Setup

Tranquility works with the Druid indexing service (http://druid.io/docs/latest/Indexing-Service.html). To get started,
you'll need an Overlord, enough Middle Managers for your realtime workload, and enough Historical nodes to receive
handoffs. You don't need any Realtime nodes, since Tranquility uses the indexing service for all of its ingestion needs.

Tranquility periodically submits new tasks to the indexing service to provide for log rotation and to support
zero-downtime configuration changes. These new tasks are typically submitted before the old ones exit, so to allow for
{'fraction_non_alphanumeric': 0.10129654845775857, 'fraction_numerical': 0.0069267657332307145, 'mean_word_length': 4.502280130293159, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 57, 'https://': 8, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6213660', 'n_tokens_mistral': 5250, 'n_tokens_neox': 5066, 'n_words': 1771}