Updated examples for bullet 1.0.0 (#38)

Author: 0aix

docs/ws/examples.md

@@ -4,14 +4,14 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.yahoo.bullet</groupId>
     <artifactId>bullet-spark-example</artifactId>
-    <version>0.0.1-SNAPSHOT</version>
+    <version>1.0.0-SNAPSHOT</version>
     <packaging>jar</packaging>
     <properties>
         <scala.version>2.11.7</scala.version>
         <scala.dep.version>2.11</scala.dep.version>
         <spark.version>2.3.0</spark.version>
-        <bullet.spark.version>0.2.2</bullet.spark.version>
-        <bullet.record.version>0.2.0</bullet.record.version>
+        <bullet.spark.version>1.0.0</bullet.spark.version>
+        <bullet.record.version>1.0.0</bullet.record.version>
     </properties>
 
     <repositories>

examples/spark/pom.xml

@@ -110,4 +110,4 @@ bullet.query.aggregation.top.k.sketch.entries: 1024
 bullet.query.aggregation.top.k.sketch.error.type: "NFN"
 bullet.result.metadata.enable: true
 # Factory class to get new BulletRecords.
-bullet.record.provider.class.name: "com.yahoo.bullet.record.SimpleBulletRecordProvider"
+bullet.record.provider.class.name: "com.yahoo.bullet.record.simple.SimpleBulletRecordProvider"

examples/spark/src/main/resources/bullet_spark_kafka_settings.yaml

@@ -13,10 +13,10 @@ import org.apache.spark.streaming.StreamingContext
 import org.apache.spark.streaming.dstream.DStream
 
 class RandomProducer extends DataProducer {
-  override def getBulletRecordStream(ssc: StreamingContext, config: BulletSparkConfig): DStream[BulletRecord] = {
+  override def getBulletRecordStream(ssc: StreamingContext, config: BulletSparkConfig): DStream[BulletRecord[_ <: java.io.Serializable]] = {
     // Bullet record input stream.
     val bulletReceiver = new RandomReceiver(config)
-    ssc.receiverStream(bulletReceiver).asInstanceOf[DStream[BulletRecord]]
+    ssc.receiverStream(bulletReceiver).asInstanceOf[DStream[BulletRecord[_ <: java.io.Serializable]]]
   }
 }
 

examples/spark/src/main/scala/com/yahoo/bullet/spark/examples/RandomProducer.scala

@@ -11,13 +11,12 @@ import java.util.Map
 import java.util.Arrays.asList
 
 import scala.util.Random
-
-import com.yahoo.bullet.record.{BulletRecord, SimpleBulletRecord}
+import com.yahoo.bullet.record.BulletRecord
+import com.yahoo.bullet.record.simple.TypedSimpleBulletRecord
 import com.yahoo.bullet.spark.utils.{BulletSparkConfig, BulletSparkLogger}
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.receiver.Receiver
 
-
 object RandomReceiver {
   // Fields in BulletRecord
   private val STRING = "uuid"
@@ -47,7 +46,7 @@ object RandomReceiver {
  * @param config The BulletSparkConfig to load settings from.
  */
 class RandomReceiver(val config: BulletSparkConfig)
-  extends Receiver[BulletRecord](StorageLevel.MEMORY_AND_DISK_SER) with BulletSparkLogger {
+  extends Receiver[BulletRecord[_ <: java.io.Serializable]](StorageLevel.MEMORY_AND_DISK_SER) with BulletSparkLogger {
   // Number of tuples to emit
   private val maxPerPeriod = 100L
   // Period in milliseconds. Default 1000 ms
@@ -100,8 +99,8 @@ class RandomReceiver(val config: BulletSparkConfig)
     randomMap
   }
 
-  private def generateRecord(): BulletRecord = {
-    val record = new SimpleBulletRecord()
+  private def generateRecord(): BulletRecord[_ <: java.io.Serializable] = {
+    val record = new TypedSimpleBulletRecord()
     val uuid = UUID.randomUUID().toString
     record.setString(RandomReceiver.STRING, uuid)
     record.setLong(RandomReceiver.LONG, generatedThisPeriod)

examples/spark/src/main/scala/com/yahoo/bullet/spark/examples/receiver/RandomReceiver.scala

@@ -1,16 +1,10 @@
 #! /bin/bash
 
 # We pass 20 and 100 to the RandomSpout, which means it generates up to 20 random records every 100 ms.
-storm jar bullet-storm-example-1.0-SNAPSHOT-jar-with-dependencies.jar \
+storm jar bullet-storm-example-1.0.0-SNAPSHOT-jar-with-dependencies.jar \
           com.yahoo.bullet.storm.Topology \
           --bullet-conf ./bullet_settings.yaml \
           --bullet-spout com.yahoo.bullet.storm.examples.RandomSpout \
-          --bullet-spout-parallelism 1 \
-          --bullet-spout-cpu-load 100.0 \
-          --bullet-spout-on-heap-memory-load 128.0 \
-          --bullet-spout-off-heap-memory-load 196.0 \
-          --bullet-spout-arg 20 \
-          --bullet-spout-arg 101 \
           -c topology.acker.executors=1 \
           -c topology.max.spout.pending=1000 \
           -c topology.backpressure.enable=false

examples/storm/bin/launch.sh

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.yahoo.bullet</groupId>
     <artifactId>bullet-storm-example</artifactId>
-    <version>1.0-SNAPSHOT</version>
+    <version>1.0.0-SNAPSHOT</version>
     <packaging>jar</packaging>
     <name>bullet-storm-example</name>
     <scm>
@@ -26,10 +26,10 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <bullet.storm.version>0.8.5</bullet.storm.version>
-        <bullet.core.version>0.6.4</bullet.core.version>
-        <bullet.record.version>0.3.0</bullet.record.version>
-        <storm.version>1.1.3</storm.version>
+        <bullet.storm.version>1.0.0</bullet.storm.version>
+        <bullet.core.version>1.0.0</bullet.core.version>
+        <bullet.record.version>1.0.0</bullet.record.version>
+        <storm.version>2.1.0</storm.version>
     </properties>
 
     <dependencies>

examples/storm/pom.xml

@@ -6,8 +6,8 @@
 package com.yahoo.bullet.storm.examples;
 
 import com.yahoo.bullet.common.BulletConfig;
-import com.yahoo.bullet.record.AvroBulletRecord;
 import com.yahoo.bullet.record.BulletRecord;
+import com.yahoo.bullet.record.avro.TypedAvroBulletRecord;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.storm.spout.SpoutOutputCollector;
 import org.apache.storm.task.TopologyContext;
@@ -144,7 +144,7 @@ private Map<String, String> makeRandomMap() {
     }
 
     private BulletRecord generateRecord() {
-        BulletRecord record = new AvroBulletRecord();
+        BulletRecord record = new TypedAvroBulletRecord();
         String uuid = UUID.randomUUID().toString();
 
         record.setString(STRING, uuid);

examples/storm/src/main/java/com/yahoo/bullet/storm/examples/RandomSpout.java

@@ -30,6 +30,13 @@ bullet.topology.join.bolt.query.post.finish.buffer.ticks: 3
 bullet.topology.join.bolt.query.pre.start.delay.ticks: 2
 bullet.topology.tick.spout.interval.ms: 100
 
+bullet.topology.bullet.spout.class.name: "com.yahoo.bullet.storm.examples.RandomSpout"
+bullet.topology.bullet.spout.args: []
+bullet.topology.bullet.spout.parallelism: 1
+bullet.topology.bullet.spout.cpu.load: 25.0
+bullet.topology.bullet.spout.memory.on.heap.load: 128.0
+bullet.topology.bullet.spout.memory.off.heap.load: 160.0
+
 # Bullet Core settings
 bullet.query.aggregation.raw.max.size: 500
 bullet.query.aggregation.max.size: 1024

examples/storm/src/main/resources/bullet_settings.yaml

@@ -31,26 +31,23 @@
     },
     {
         "name": "subtypes",
-        "type": "MAP",
-        "subtype": "STRING",
+        "type": "STRING_MAP",
         "description": "Contains two keys whose values are randomly chosen from: foo, bar, baz, qux, quux, norf",
-        "enumerations": [
+        "subFields": [
             {"name": "field_A", "description": "Value randomly chosen from: foo, bar, baz, qux, quux, norf"},
             {"name": "field_B", "description": "Value randomly chosen from: foo, bar, baz, qux, quux, norf"}
         ]
     },
     {
         "name": "tags",
-        "type": "MAP",
-        "subtype": "BOOLEAN",
+        "type": "BOOLEAN_MAP",
         "description": "Contains four keys which are four fragments of the uuid. The values are randomly generated boolean values from Random#nextBoolean"
     },
     {
         "name": "stats",
-        "type": "MAP",
-        "subtype": "LONG",
+        "type": "LONG_MAP",
         "description": "This map contains some numeric information such as the current number of periods etc.",
-        "enumerations": [
+        "subFields": [
             {"name": "period_count", "description": "The period in which this record was generated"},
             {"name": "record_number", "description": "A monotonically increasing id for the record. There may be gaps in the id but if the data generation has kept up with your maximum tuples per period, this is the nth tuple generated"},
             {"name": "timestamp", "description": "The ms time when this record was generated"},
@@ -59,8 +56,7 @@
     },
     {
         "name": "classifiers",
-        "type": "LIST",
-        "subtype": "MAP",
+        "type": "STRING_MAP_LIST",
         "description": "This contains two maps, each with: field_A and field_B whose values are randomly chosen from: foo, bar, baz, qux, quux, norf"
     }
 ]