快速学习-Druid数据摄入之Tranquility

<dependencies>
        <dependency>
            <groupId>io.druid</groupId>
            <artifactId>tranquility-core_2.11</artifactId>
            <version>0.8.2</version>
        </dependency>
        <dependency>
            <groupId>com.metamx</groupId>
            <artifactId>java-util</artifactId>
            <version>1.3.2</version>
        </dependency>
    </dependencies>

{
  "dataSources": [
    {
      "spec": {
        "dataSchema": {
          "dataSource": "wikipedia02",
          "parser": {
            "type": "string",
            "parseSpec": {
              "format": "json",
              "timestampSpec": {
                "column": "timestamp",
                "format": "auto"
              },
              "dimensionsSpec": {
                "dimensions": [
                  "page",
                  "language",
                  "user",
                  "unpatrolled",
                  "newPage",
                  "robot",
                  "anonymous",
                  "namespace",
                  "continent",
                  "country",
                  "region",
                  "city"
                ],
                "dimensionExclusions": [],
                "spatialDimensions": []
              }
            }
          },
          "metricsSpec": [
            {
              "type": "count",
              "name": "count"
            },
            {
              "type": "doubleSum",
              "name": "added",
              "fieldName": "added"
            },
            {
              "type": "doubleSum",
              "name": "deleted",
              "fieldName": "deleted"
            },
            {
              "type": "doubleSum",
              "name": "delta",
              "fieldName": "delta"
            }
          ],
          "granularitySpec": {
            "type": "uniform",
            "segmentGranularity": "DAY",
            "queryGranularity": "NONE"
          }
        },
        "tuningConfig": {
          "type": "realtime",
          "maxRowsInMemory": 100000,
          "intermediatePersistPeriod": "PT10m",
          "windowPeriod": "PT10M"
        }
      }
    }
  ],
  "properties": {
    "zookeeper.connect": "hadoop102:2181",
    "druid.selectors.indexing.serviceName": "druid/overlord",
    "druid.discovery.curator.path": "/druid/discovery",
    "druidBeam.taskLocator": "overlord",
    "druidBeam.overlordPollPeriod": "PT5S"
  }
}

public class JavaExample {
    private static final Logger log = new Logger(JavaExample.class);

    public static void main(String[] args)
    {
        // Read config from "example.json" on the classpath.
        final InputStream configStream = JavaExample.class.getClassLoader().getResourceAsStream("example.json");
        final TranquilityConfig<PropertiesBasedConfig> config = TranquilityConfig.read(configStream);
        final DataSourceConfig<PropertiesBasedConfig> wikipediaConfig = config.getDataSource("wikipedia02");
        final Tranquilizer<Map<String, Object>> sender = DruidBeams.fromConfig(wikipediaConfig)
                .buildTranquilizer(wikipediaConfig.tranquilizerBuilder());

        sender.start();

        try {
            // Send 10000 objects

            for (int i = 0; i < 10; i++) {
                // Build a sample event to send; make sure we use a current date
                final Map<String, Object> obj = ImmutableMap.<String, Object>of(
                        "timestamp", new DateTime().toString(),
                        "page", "foo",
                        "added", i
                );

                // Asynchronously send event to Druid:
                sender.send(obj).addEventListener(
                        new FutureEventListener<BoxedUnit>()
                        {
                            @Override
                            public void onSuccess(BoxedUnit value)
                            {
                                log.info("Sent message: %s", obj);
                            }

                            @Override
                            public void onFailure(Throwable e)
                            {
                                if (e instanceof MessageDroppedException) {
                                    log.warn(e, "Dropped message: %s", obj);
                                } else {
                                    log.error(e, "Failed to send message: %s", obj);
                                }
                            }
                        }
                );
            }
        }
        finally {
            sender.flush();
            sender.stop();
        }
    }
}

object ScalaExample extends Logging
{
  def main(args: Array[String]) {
    // Read config from "example.json" on the classpath.
    val configStream = getClass.getClassLoader.getResourceAsStream("example.json")
    val config: TranquilityConfig[PropertiesBasedConfig] = TranquilityConfig.read(configStream)
    val wikipediaConfig: DataSourceConfig[PropertiesBasedConfig] = config.getDataSource("wikipedia03")
    val sender: Tranquilizer[java.util.Map[String, AnyRef]] = DruidBeams
      .fromConfig(config.getDataSource("wikipedia03"))
      .buildTranquilizer(wikipediaConfig.tranquilizerBuilder())

    sender.start()

    try {
      // Send 10000 objects.

      for (i <- 0 until 10) {
        val obj = Map[String, AnyRef](
          "timestamp" -> new DateTime().toString,
          "page" -> "foo",
          "added" -> Int.box(i)
        )

        // Asynchronously send event to Druid:
        sender.send(obj.asJava) respond {
          case Return(_) =>
            log.info("Sent message: %s", obj)

          case Throw(e: MessageDroppedException) =>
            log.warn(e, "Dropped message: %s", obj)

          case Throw(e) =>
            log.error(e, "Failed to send message: %s", obj)
        }
      }
    }
    finally {
      sender.flush()
      sender.stop()
    }
  }
}

<dependencies>
        <dependency>
            <groupId>io.druid</groupId>
            <artifactId>tranquility-core_2.11</artifactId>
            <version>0.8.2</version>
        </dependency>
        <dependency>
            <groupId>com.metamx</groupId>
            <artifactId>java-util</artifactId>
            <version>1.3.2</version>
        </dependency>
<dependency>
            <groupId>io.druid</groupId>
            <artifactId>tranquility-spark_2.11</artifactId>
            <version>0.8.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>
    </dependencies>

class SimpleEventBeamFactory extends BeamFactory[SimpleEvent]
{
  // Return a singleton, so the same connection is shared across all tasks in the same JVM.
  def makeBeam: Beam[SimpleEvent] = SimpleEventBeamFactory.BeamInstance
}

object SimpleEventBeamFactory
{
  val BeamInstance: Beam[SimpleEvent] = {
    // Tranquility uses ZooKeeper (through Curator framework) for coordination.
    val curator = CuratorFrameworkFactory.newClient(
      "localhost:2181",
      new BoundedExponentialBackoffRetry(100, 3000, 5)
    )
    curator.start()

    val indexService = "druid/overlord" // Your overlord's druid.service, with slashes replaced by colons.
    val discoveryPath = "/druid/discovery"     // Your overlord's druid.discovery.curator.path
    val dataSource = "foo"
    val dimensions = IndexedSeq("bar")
    val aggregators = Seq(new LongSumAggregatorFactory("baz", "baz"))
    val isRollup = true

    // Expects simpleEvent.timestamp to return a Joda DateTime object.
    DruidBeams
      .builder((simpleEvent: SimpleEvent) => simpleEvent.timestamp)
      .curator(curator)
      .discoveryPath(discoveryPath)
      .location(DruidLocation(indexService, dataSource))
      .rollup(DruidRollup(SpecificDruidDimensions(dimensions), aggregators, QueryGranularities.MINUTE, isRollup))
      .tuning(
        ClusteredBeamTuning(
          segmentGranularity = Granularity.HOUR,
          windowPeriod = new Period("PT10M"),
          partitions = 1,
          replicants = 1
        )
      )
      .buildBeam()
  }
}

// Add this import to your Spark job to be able to propagate events from any RDD to Druid
import com.metamx.tranquility.spark.BeamRDD._

// Now given a Spark DStream, you can send events to Druid.
dstream.foreachRDD(rdd => rdd.propagate(new SimpleEventBeamFactory))