phase 2 template

Author: iamjagdeesh

README.md renamed to phase_1/README.md

@@ -0,0 +1 @@
+sbt.version=1.2.6

build.sbt renamed to phase_1/build.sbt

@@ -0,0 +1,4 @@
+/.idea/
+/spark-warehouse/
+/target/
+/test/

exampleanswer renamed to phase_1/exampleanswer

@@ -0,0 +1,138 @@
+# CSE512-Project-Hotspot-Analysis-Template
+#### Version history
+v1.1, Nov 16, Fix a bug in "Entrace.scala"
+v1.0, Nov 13, Initial version
+
+
+## Requirement
+
+In this phase, you are required to do spatial hot spot analysis. In particular, you need to complete two different hot spot analysis tasks 
+
+
+### Hot zone analysis
+This task will needs to perform a range join operation on a rectangle datasets and a point dataset. For each rectangle, the number of points located within the rectangle will be obtained. The hotter rectangle means that it include more points. So this task is to calculate the hotness of all the rectangles. 
+
+### Hot cell analysis
+
+#### Description
+This task will focus on applying spatial statistics to spatio-temporal big data in order to identify statistically significant spatial hot spots using Apache Spark. The topic of this task is from ACM SIGSPATIAL GISCUP 2016.
+
+The Problem Definition page is here: [http://sigspatial2016.sigspatial.org/giscup2016/problem](http://sigspatial2016.sigspatial.org/giscup2016/problem) 
+
+The Submit Format page is here: [http://sigspatial2016.sigspatial.org/giscup2016/submit](http://sigspatial2016.sigspatial.org/giscup2016/submit)
+
+#### Special requirement (different from GIS CUP)
+As stated in the Problem Definition page, in this task, you are asked to implement a Spark program to calculate the Getis-Ord statistic of NYC Taxi Trip datasets. We call it "**Hot cell analysis**"
+
+To reduce the computation power need，we made the following changes:
+
+1. The input will be a monthly taxi trip dataset from 2009 - 2012. For example, "yellow\_tripdata\_2009-01\_point.csv", "yellow\_tripdata\_2010-02\_point.csv".
+2. Each cell unit size is 0.01 * 0.01 in terms of latitude and longitude degrees.
+3. You should use 1 day as the Time Step size. The first day of a month is step 1. Every month has 31 days.
+4. You only need to consider Pick-up Location.
+5. We don't use Jaccard similarity to check your answer.
+However, you don't need to worry about how to decide the cell coordinates because the code template generated cell coordinates. You just need to write the rest of the task.
+
+## Coding template specification
+
+### Input parameters
+
+1. Output path (Mandatory)
+2. Task name: "hotzoneanalysis" or "hotcellanalysis"
+3. Task parameters: (1) Hot zone (2 parameters): nyc taxi data path, zone path(2) Hot cell (1 parameter): nyc taxi data path
+
+Example
+```
+test/output hotzoneanalysis src/resources/point-hotzone.csv src/resources/zone-hotzone.csv hotcellanalysis src/resources/yellow_trip_sample_100000.csv
+```
+
+Note: 
+
+1. The number/order of tasks do not matter.
+2. But, the first 7 of our final test cases will be hot zone analysis, the last 8 will be hot cell analysis.
+
+
+
+
+### Input data format
+The main function/entrace is "cse512.Entrance" scala file.
+
+1. Point data: the input point dataset is the pickup point of New York Taxi trip datasets. The data format of this phase is the original format of NYC taxi trip which is different from the previous phase. But the coding template already parsed it for you. Find the data from our asu google drive shared folder: [https://drive.google.com/drive/folders/1W4GLKNsGlgXp7fHtDlhHEBdLVw_IuAXh?usp=sharing](https://drive.google.com/drive/folders/1W4GLKNsGlgXp7fHtDlhHEBdLVw_IuAXh?usp=sharing)
+
+2. Zone data (only for hot zone analysis): at "src/resources/zone-hotzone" of the template
+
+#### Hot zone analysis
+The input point data can be any small subset of NYC taxi dataset.
+
+#### Hot cell analysis
+The input point data is a monthly NYC taxi trip dataset (2009-2012) like "yellow\_tripdata\_2009-01\_point.csv"
+
+### Output data format
+
+#### Hot zone analysis
+All zones with their count, sorted by "rectangle" string in an ascending order. 
+
+```
+"-73.795658,40.743334,-73.753772,40.779114",1
+"-73.797297,40.738291,-73.775740,40.770411",1
+"-73.832707,40.620010,-73.746541,40.665414",20
+```
+
+
+#### Hot cell analysis
+The coordinates of top 50 hotest cells sorted by their G score in a descending order. Note, DO NOT OUTPUT G score.
+
+```
+-7399,4075,15
+-7399,4075,29
+-7399,4075,22
+```
+### Example answers
+An example input and answer are put in "testcase" folder of the coding template
+
+
+## Where you need to change
+DO NOT DELETE any existing code in the coding template unless you see this "YOU NEED TO CHANGE THIS PART"
+
+### Hot zone analysis
+
+In the code template,
+
+1. You need to change "**HotzoneAnalysis.scala** and **HotzoneUtils.scala**".
+2. The coding template has loaded the data and wrote the first step, range join query, for you. Please finish the rest of the task.
+3. The output DataFrame should be sorted by you according to "rectangle" string.
+
+### Hot cell analysis
+In the code template,
+
+1. You need to change "**HotcellAnalysis.scala** and **HotcellUtils.scala**".
+2. The coding template has loaded the data and decided the cell coordinate, x, y, z and their min and max. Please finish the rest of the task.
+3. The output DataFrame should be sorted by you according to G-score. The coding template will take the first 50 to output. DO NOT OUTPUT G-score.
+
+
+## Submission
+### Submission files
+1. Submit your project jar package.
+2. Submit your project source code onto Blackboard and follow the submission instruction on BB.
+3. Note that: you need to make sure your code can compile and package by entering ```sbt clean assembly```. We will run the compiled package on our cluster directly using "spark-submit" with parameters. If your code cannot compile and package, you will not receive any points.
+
+## Tips (Optional)
+This section is same with that in Phase 1.
+### How to debug your code in IDE
+
+If you are using the Scala template
+
+1. Use IntelliJ Idea with Scala plug-in or any other Scala IDE.
+2. Replace the logic of User Defined Functions ST\_Contains and ST\_Within in SpatialQuery.scala.
+3. Append ```.master("local[*]")``` after ```.config("spark.some.config.option", "some-value")``` to tell IDE the master IP is localhost.
+3. In some cases, you may need to go to "build.sbt" file and change ```% "provided"``` to ```% "compile"``` in order to debug your code in IDE
+4. Run your code in IDE
+5. **You must revert Step 3 and 4 above and recompile your code before use spark-submit!!!**
+
+### How to submit your code to Spark
+If you are using the Scala template
+
+1. Go to project root folder
+2. Run ```sbt clean assembly```. You may need to install sbt in order to run this command.
+3. Find the packaged jar in "./target/scala-2.11/CSE512-Project-Hotspot-Analysis-Template-assembly-0.1.0.jar"
+4. Submit the jar to Spark using Spark command "./bin/spark-submit". A pseudo code example: ```./bin/spark-submit ~/GitHub/CSE512-Project-Hotspot-Analysis-Template/target/scala-2.11/CSE512-Project-Hotspot-Analysis-Template-assembly-0.1.0.jar test/output hotzoneanalysis src/resources/point-hotzone.csv src/resources/zone-hotzone.csv hotcellanalysis src/resources/yellow_tripdata_2009-01_point.csv```

exampleinput renamed to phase_1/exampleinput

@@ -0,0 +1,25 @@
+import sbt.Keys.{libraryDependencies, scalaVersion, version}
+
+
+lazy val root = (project in file(".")).
+  settings(
+    name := "CSE512-Hotspot-Analysis-Template",
+
+    version := "0.1.0",
+
+    scalaVersion := "2.11.11",
+
+    organization  := "org.datasyslab",
+
+    publishMavenStyle := true,
+
+    mainClass := Some("cse512.Entrance")
+  )
+
+libraryDependencies ++= Seq(
+  "org.apache.spark" %% "spark-core" % "2.2.0" % "provided",
+  "org.apache.spark" %% "spark-sql" % "2.2.0" % "provided",
+  "org.scalatest" %% "scalatest" % "2.2.4" % "test",
+  "org.specs2" %% "specs2-core" % "2.4.16" % "test",
+  "org.specs2" %% "specs2-junit" % "2.4.16" % "test"
+)

project/.gitignore renamed to phase_1/project/.gitignore

@@ -0,0 +1,3 @@
+/target/
+/project/
+*.properties

phase_1/project/build.properties

@@ -0,0 +1 @@
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.5")

project/plugins.sbt renamed to phase_1/project/plugins.sbt

@@ -0,0 +1,63 @@
+package cse512
+
+import org.apache.log4j.{Level, Logger}
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
+
+object Entrance extends App {
+  Logger.getLogger("org.spark_project").setLevel(Level.WARN)
+  Logger.getLogger("org.apache").setLevel(Level.WARN)
+  Logger.getLogger("akka").setLevel(Level.WARN)
+  Logger.getLogger("com").setLevel(Level.WARN)
+
+  override def main(args: Array[String]) {
+    val spark = SparkSession
+      .builder()
+      .appName("CSE512-HotspotAnalysis-MYGROUPNAME") // YOU NEED TO CHANGE YOUR GROUP NAME
+      .config("spark.some.config.option", "some-value")//.master("local[*]")
+      .getOrCreate()
+
+    paramsParser(spark, args)
+
+  }
+  
+  private def paramsParser(spark: SparkSession, args: Array[String]): Unit = {
+    var paramOffset = 1
+    var currentQueryParams = ""
+    var currentQueryName = ""
+    var currentQueryIdx = -1
+
+    while (paramOffset <= args.length) {
+      if (paramOffset == args.length || args(paramOffset).toLowerCase.contains("analysis")) {
+        // Turn in the previous query
+        if (currentQueryIdx != -1) queryLoader(spark, currentQueryName, currentQueryParams, args(0) + currentQueryIdx)
+
+        // Start a new query call
+        if (paramOffset == args.length) return
+
+        currentQueryName = args(paramOffset)
+        currentQueryParams = ""
+        currentQueryIdx = currentQueryIdx + 1
+      }
+      else {
+        // Keep appending query parameters
+        currentQueryParams = currentQueryParams + args(paramOffset) + " "
+      }
+      paramOffset = paramOffset + 1
+    }
+  }
+
+  private def queryLoader(spark: SparkSession, queryName: String, queryParams: String, outputPath: String) {
+    val queryParam = queryParams.split(" ")
+    if (queryName.equalsIgnoreCase("hotcellanalysis")) {
+      if (queryParam.length != 1) throw new ArrayIndexOutOfBoundsException("[CSE512] Query " + queryName + " needs 1 parameters but you entered " + queryParam.length)
+      HotcellAnalysis.runHotcellAnalysis(spark, queryParam(0)).limit(50).write.mode(SaveMode.Overwrite).csv(outputPath)
+    }
+    else if (queryName.equalsIgnoreCase("hotzoneanalysis")) {
+      if (queryParam.length != 2) throw new ArrayIndexOutOfBoundsException("[CSE512] Query " + queryName + " needs 2 parameters but you entered " + queryParam.length)
+      HotzoneAnalysis.runHotZoneAnalysis(spark, queryParam(0), queryParam(1)).write.mode(SaveMode.Overwrite).csv(outputPath)
+    }
+    else {
+      throw new NoSuchElementException("[CSE512] The given query name " + queryName + " is wrong. Please check your input.")
+    }
+  }
+}

src/main/scala/cse512/SparkSQLExample.scala renamed to phase_1/src/main/scala/cse512/SparkSQLExample.scala

@@ -0,0 +1,49 @@
+package cse512
+
+import org.apache.log4j.{Level, Logger}
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.functions.udf
+import org.apache.spark.sql.functions._
+
+object HotcellAnalysis {
+  Logger.getLogger("org.spark_project").setLevel(Level.WARN)
+  Logger.getLogger("org.apache").setLevel(Level.WARN)
+  Logger.getLogger("akka").setLevel(Level.WARN)
+  Logger.getLogger("com").setLevel(Level.WARN)
+
+def runHotcellAnalysis(spark: SparkSession, pointPath: String): DataFrame =
+{
+  // Load the original data from a data source
+  var pickupInfo = spark.read.format("com.databricks.spark.csv").option("delimiter",";").option("header","false").load(pointPath);
+  pickupInfo.createOrReplaceTempView("nyctaxitrips")
+  pickupInfo.show()
+
+  // Assign cell coordinates based on pickup points
+  spark.udf.register("CalculateX",(pickupPoint: String)=>((
+    HotcellUtils.CalculateCoordinate(pickupPoint, 0)
+    )))
+  spark.udf.register("CalculateY",(pickupPoint: String)=>((
+    HotcellUtils.CalculateCoordinate(pickupPoint, 1)
+    )))
+  spark.udf.register("CalculateZ",(pickupTime: String)=>((
+    HotcellUtils.CalculateCoordinate(pickupTime, 2)
+    )))
+  pickupInfo = spark.sql("select CalculateX(nyctaxitrips._c5),CalculateY(nyctaxitrips._c5), CalculateZ(nyctaxitrips._c1) from nyctaxitrips")
+  var newCoordinateName = Seq("x", "y", "z")
+  pickupInfo = pickupInfo.toDF(newCoordinateName:_*)
+  pickupInfo.show()
+
+  // Define the min and max of x, y, z
+  val minX = -74.50/HotcellUtils.coordinateStep
+  val maxX = -73.70/HotcellUtils.coordinateStep
+  val minY = 40.50/HotcellUtils.coordinateStep
+  val maxY = 40.90/HotcellUtils.coordinateStep
+  val minZ = 1
+  val maxZ = 31
+  val numCells = (maxX - minX + 1)*(maxY - minY + 1)*(maxZ - minZ + 1)
+
+  // YOU NEED TO CHANGE THIS PART
+
+  return pickupInfo // YOU NEED TO CHANGE THIS PART
+}
+}

src/main/scala/cse512/SpatialQuery.scala renamed to phase_1/src/main/scala/cse512/SpatialQuery.scala

@@ -0,0 +1,51 @@
+package cse512
+
+import java.sql.Timestamp
+import java.text.SimpleDateFormat
+import java.util.Calendar
+
+object HotcellUtils {
+  val coordinateStep = 0.01
+
+  def CalculateCoordinate(inputString: String, coordinateOffset: Int): Int =
+  {
+    // Configuration variable:
+    // Coordinate step is the size of each cell on x and y
+    var result = 0
+    coordinateOffset match
+    {
+      case 0 => result = Math.floor((inputString.split(",")(0).replace("(","").toDouble/coordinateStep)).toInt
+      case 1 => result = Math.floor(inputString.split(",")(1).replace(")","").toDouble/coordinateStep).toInt
+      // We only consider the data from 2009 to 2012 inclusively, 4 years in total. Week 0 Day 0 is 2009-01-01
+      case 2 => {
+        val timestamp = HotcellUtils.timestampParser(inputString)
+        result = HotcellUtils.dayOfMonth(timestamp) // Assume every month has 31 days
+      }
+    }
+    return result
+  }
+
+  def timestampParser (timestampString: String): Timestamp =
+  {
+    val dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss")
+    val parsedDate = dateFormat.parse(timestampString)
+    val timeStamp = new Timestamp(parsedDate.getTime)
+    return timeStamp
+  }
+
+  def dayOfYear (timestamp: Timestamp): Int =
+  {
+    val calendar = Calendar.getInstance
+    calendar.setTimeInMillis(timestamp.getTime)
+    return calendar.get(Calendar.DAY_OF_YEAR)
+  }
+
+  def dayOfMonth (timestamp: Timestamp): Int =
+  {
+    val calendar = Calendar.getInstance
+    calendar.setTimeInMillis(timestamp.getTime)
+    return calendar.get(Calendar.DAY_OF_MONTH)
+  }
+
+  // YOU NEED TO CHANGE THIS PART
+}

src/resources/arealm10000.csv renamed to phase_1/src/resources/arealm10000.csv

@@ -0,0 +1,37 @@
+package cse512
+
+import org.apache.log4j.{Level, Logger}
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
+
+object HotzoneAnalysis {
+
+  Logger.getLogger("org.spark_project").setLevel(Level.WARN)
+  Logger.getLogger("org.apache").setLevel(Level.WARN)
+  Logger.getLogger("akka").setLevel(Level.WARN)
+  Logger.getLogger("com").setLevel(Level.WARN)
+
+  def runHotZoneAnalysis(spark: SparkSession, pointPath: String, rectanglePath: String): DataFrame = {
+
+    var pointDf = spark.read.format("com.databricks.spark.csv").option("delimiter",";").option("header","false").load(pointPath);
+    pointDf.createOrReplaceTempView("point")
+
+    // Parse point data formats
+    spark.udf.register("trim",(string : String)=>(string.replace("(", "").replace(")", "")))
+    pointDf = spark.sql("select trim(_c5) as _c5 from point")
+    pointDf.createOrReplaceTempView("point")
+
+    // Load rectangle data
+    val rectangleDf = spark.read.format("com.databricks.spark.csv").option("delimiter","\t").option("header","false").load(rectanglePath);
+    rectangleDf.createOrReplaceTempView("rectangle")
+
+    // Join two datasets
+    spark.udf.register("ST_Contains",(queryRectangle:String, pointString:String)=>(HotzoneUtils.ST_Contains(queryRectangle, pointString)))
+    val joinDf = spark.sql("select rectangle._c0 as rectangle, point._c5 as point from rectangle,point where ST_Contains(rectangle._c0,point._c5)")
+    joinDf.createOrReplaceTempView("joinResult")
+
+    // YOU NEED TO CHANGE THIS PART
+
+    return joinDf // YOU NEED TO CHANGE THIS PART
+  }
+
+}

src/resources/zcta10000.csv renamed to phase_1/src/resources/zcta10000.csv

@@ -0,0 +1,12 @@
+package cse512
+
+object HotzoneUtils {
+
+  def ST_Contains(queryRectangle: String, pointString: String ): Boolean = {
+    // YOU NEED TO CHANGE THIS PART
+    return true // YOU NEED TO CHANGE THIS PART
+  }
+
+  // YOU NEED TO CHANGE THIS PART
+
+}

phase_2/.gitignore

@@ -0,0 +1 @@
+/yellow_tripdata_2009-01_point.csv