Changes after dry run on the VM

FinanceStreamSimulator/build/resources/main/application.properties

@@ -1,5 +1,5 @@
 serverUrl=http://localhost:9000
-geodeUrl=http://geode-server:8080
-numberOfMessages=1000
+geodeUrl=http://localhost:8888
+numberOfMessages=500
 delayInMs=300
 symbol=TSLA

FinanceStreamSimulator/src/main/resources/application.properties

@@ -1,5 +1,5 @@
 serverUrl=http://localhost:9000
-geodeUrl=http://geode-server:8080
-numberOfMessages=1000
+geodeUrl=http://localhost:8888
+numberOfMessages=500
 delayInMs=300
 symbol=TSLA

FinanceStreamSimulator/startSimulator.sh

@@ -1,9 +1,5 @@
 #!/bin/bash
 
-SIMULATOR_CONTAINER=`docker ps |grep -i java:7 | awk '{print $1}'`
-
-echo Using Simulator container ID: $SIMULATOR_CONTAINER
-
-docker exec $SIMULATOR_CONTAINER FinanceStreamSimulator/gradlew -p FinanceStreamSimulator run
+gradle -p run
 
 

SpringXD.md

@@ -91,7 +91,8 @@ xd:>stream all destroy
 
 ## Final step
 
-* Before moving the to next lab run the script `stream-create.sh` from `$PROJECT/streaming/`
+* Before moving the to next lab run the script
+`stream-create.sh` from `$PROJECT/streaming/`
 * After the script execute, access the following from a browser:
   * http://192.168.56.10:9393/admin-ui/#/streams/definitions
   * http://192.168.56.10:9393/admin-ui/#/streams/create

StocksSpark/build.sbt

@@ -1,13 +1,15 @@
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.13.0")
+
 name := "StocksSpark"
 version := "1.0"
 scalaVersion := "2.10.5"
 
 // additional libraries
 libraryDependencies ++= Seq(
   "org.apache.spark" %% "spark-core" % "1.3.1" % "provided",
-  "org.apache.spark" %% "spark-sql" % "1.3.1",
-  "org.apache.spark" %% "spark-streaming" % "1.3.1",
-  "org.apache.spark" %% "spark-mllib" % "1.3.1",
+  "org.apache.spark" %% "spark-sql" % "1.3.1" % "provided",
+  "org.apache.spark" %% "spark-streaming" % "1.3.1" % "provided",
+  "org.apache.spark" %% "spark-mllib" % "1.3.1" % "provided",
   "org.apache.geode" % "gemfire-core" % "1.0.0-incubating-SNAPSHOT" excludeAll(ExclusionRule(organization = "org.jboss.netty")),
   "io.pivotal.gemfire.spark" % "gemfire-spark-connector_2.10" % "0.5.0" excludeAll(ExclusionRule(organization = "org.apache.spark")),
   "org.scalatest" %% "scalatest" % "2.2.1" % "test"
@@ -16,5 +18,8 @@ libraryDependencies ++= Seq(
 resolvers ++= Seq(
   "Apache Software Foundation" at "http://repository.apache.org/snapshots/",
   "Local Maven" at Path.userHome.asFile.toURI.toURL + ".m2/repository",
-  Resolver.sonatypeRepo("public")
+  Resolver.sonatypeRepo("public"),
+  Resolver.url("bintray-sbt-plugins", url("http://dl.bintray.com/sbt/sbt-plugin-releases"))(Resolver.ivyStylePatterns)
 )
+
+//assemblyJarName in assembly := s"${name.value}-${version.value}"

streaming/stream-create.xd

@@ -1,14 +1,13 @@
 admin config server http://localhost:9393
 
-stream create process_sink --definition "queue:stocks >  transform --script='file:streaming/transform.groovy' | object-to-json | gemfire-json-server --useLocator=true --host=localhost --port=10334 --regionName=Stocks --keyExpression=payload.getField('entryTimestamp')" --deploy
+stream create process_sink --definition "queue:stocks >  transform --script='file:./transform.groovy' | object-to-json | gemfire-json-server --useLocator=true --host=localhost --port=10334 --regionName=Stocks --keyExpression=payload.getField('entryTimestamp')" --deploy
 
 stream create yahoo_finance_source --definition "trigger --cron='* * 7-13 * * MON-FRI' | http-client --url='''https://query.yahooapis.com/v1/public/yql?q=select Symbol, LastTradeDate, LastTradeTime, LastTradePriceOnly, DaysHigh, DaysLow, Open from yahoo.finance.quotes where symbol in (\"TSLA\")&format=json&env=store://datatables.org/alltableswithkeys''' --httpMethod=GET | splitter --expression=#jsonPath(payload,'$.query.results.quote')  > queue:stocks" --deploy
 
 stream create http_source --definition "http --port=9000 | splitter --expression=#jsonPath(payload,'$') > queue:stocks" --deploy
 
-stream create --name r_process --definition "tap:stream:process_sink.transform > r-parsing: object-to-json | shell --command='Rscript streaming/tech_indicators.R' | formatting: splitter --expression=#jsonPath(payload,'$') | filter --expression=#jsonPath(payload,'$.rsi').indexOf('NaN')==-1 | object-to-json | gemfire-json-server --useLocator=true --host=locator --port=10334 --regionName=TechIndicators --keyExpression=payload.getField('entryTimestamp')" --deploy
+stream create --name r_process --definition "tap:stream:process_sink.transform > r-parsing: object-to-json | shell --command='Rscript ./tech_indicators.R' | formatting: splitter --expression=#jsonPath(payload,'$') | filter --expression=#jsonPath(payload,'$.rsi').indexOf('NaN')==-1 | object-to-json | gemfire-json-server --useLocator=true --host=locator --port=10334 --regionName=TechIndicators --keyExpression=payload.getField('entryTimestamp')" --deploy
 
-stream create --name prediction --definition "tap:stream:r_process.object-to-json > shell --command='./evaluate.sh' | gemfire-json-server --regionName=Predictions --host=localhost --port=10334 --useLocator=true --keyExpression=payload.getField('entryTimestamp')" --deploy
-
-stream create --name training --definition "trigger --fixedDelay=300 | shell --command='./train.sh'" --deploy
+stream create --name prediction --definition "tap:stream:r_process.object-to-json > shell --command='../evaluate.sh' | gemfire-json-server --regionName=Predictions --host=localhost --port=10334 --useLocator=true --keyExpression=payload.getField('entryTimestamp')" --deploy
 
+# stream create --name training --definition "trigger --fixedDelay=300 | shell --command='../train.sh'" --deploy

streaming/tech_indicators.R

@@ -14,7 +14,7 @@ while(TRUE) {
   streamRow <- fromJSON(line)
 
 
-  historical <- getURL(paste0('http://geode-server:8080/gemfire-api/v1/queries/adhoc?q=SELECT%20DISTINCT%20*%20FROM%20/Stocks%20s%20WHERE%20s.entryTimestamp%20<%20',streamRow$entryTimestamp,'l%20ORDER%20BY%20s.entryTimestamp%20desc%20LIMIT%20100'))
+  historical <- getURL(paste0('http://localhost:8888/gemfire-api/v1/queries/adhoc?q=SELECT%20DISTINCT%20*%20FROM%20/Stocks%20s%20WHERE%20s.entryTimestamp%20<%20',streamRow$entryTimestamp,'l%20ORDER%20BY%20s.entryTimestamp%20desc%20LIMIT%20100'))
 
 
   historicalSet <- fromJSON(historical)