Programming Exercise from 2018 with Hart

CityOfNewYork.scala

/**
  * Created by: Angel Ortega
  * Date: 7/23/18
  */
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions.{count, max, sum}


object CityOfNewYorkChallenge extends App {
  val spark = SparkSession.builder()
    .master("local")
    .appName("City of New York")
    .getOrCreate()
  val sc = spark.sparkContext
  import spark.sqlContext.implicits._


  val newYorkDataSetDF = spark.read
    .option("header", "true")
    .option("mode", "FAILFAST")
    .option("inferSchema", "true")
    .csv("./src/main/resources/New_York_City_Leading_Causes_of_Death_2.csv")
    .cache()

  // Answer to Question 1
  val targetDemographic = newYorkDataSetDF
    .select("*")
    .where($"Sex".equalTo("F") &&
      $"`Race Ethnicity`".equalTo("White Non-Hispanic") && (
        $"`Leading Cause`".equalTo("Influenza (Flu) and Pneumonia (J09-J18)") ||
        $"`Leading Cause`".equalTo("Cerebrovascular Disease (Stroke: I60-I69)") ||
        $"`Leading Cause`".equalTo("Chronic Lower Respiratory Diseases (J40-J47)"))
    )



  // Answer to Question 2
  val topThreeDeadlyYears = newYorkDataSetDF
    .groupBy($"Year")
    .agg(
      sum($"Deaths".cast("int")).alias("total_deaths")
    )
    .orderBy($"total_deaths".desc)
    .limit(3)
    .cache()


  // Question 3

  val leadingCauseIntermediate = newYorkDataSetDF.as("ny")
    .join(
      topThreeDeadlyYears.as("deadly_years"), $"ny.Year" === $"deadly_years.Year")
    .select($"ny.*")

  val topCauses = leadingCauseIntermediate
    .groupBy("Leading Cause")
    .agg(count($"Leading Cause").alias("top_causes"))
    .orderBy($"top_causes".desc)
    .limit(3)


  // Answer to Question 3

  val mostSusceptibleRace = topCauses.as("top_3_causes").join(
    leadingCauseIntermediate.as("deadly_years"),
    $"top_3_causes.`Leading Cause`" === $"deadly_years.`Leading Cause`")
    .select($"deadly_years.Year", $"deadly_years.*", $"deadly_years.`Race Ethnicity`")
    .groupBy($"Year", $"`Leading Cause`")
    .agg(
      max($"Race Ethnicity").alias("most_susceptible_race")
    )
  targetDemographic.write
    .mode(SaveMode.Overwrite)
    .format("csv")
    .option("path", "./output/results1.csv")
    .save()

  topThreeDeadlyYears.write
    .format("csv")
    .mode(SaveMode.Overwrite)
    .option("path", "./output/results2.csv")
    .save()

  mostSusceptibleRace.toDF().write
    .format("csv")
    .mode(SaveMode.Overwrite)
    .option("path", "./output/results3.csv")
    .save()

}

CommonFriendsChallenge.scala

import org.apache.spark.sql.{SparkSession}


object CommonFriendsChallenge extends App {

  val spark = SparkSession.builder()
    .master("local")
    .appName("Common Friends")
    .getOrCreate()

  val sc = spark.sparkContext

  val friendsSmall = sc.textFile("./src/main/resources/friends_small_data.txt")
    .flatMap(line => {
      val arr = line.split("->")
      val person = arr(0).trim()
      val friends = arr(1).trim().split(",")
      for (friend <- friends) yield (friend, person)
    }).cache()
  val commonFriendsSmall = friendsSmall.groupByKey().map {
    case (k, v) => v.mkString("(", ",", ")") + "-> " + k.mkString(",")
  }

  commonFriendsSmall.saveAsTextFile("./output/results_friends_small.txt")

  val friendsLarge = sc.textFile("./src/main/resources/friends_data.txt")
    .flatMap(line => {
      val arr = line.split("->")
      val person = arr(0).trim()
      val friends = arr(1).trim().split(",")
      for (friend <- friends) yield (friend, person)
    })
    .cache()

  val commonFriendsLarge = friendsLarge.groupByKey().map {
    case (k, v) => v.mkString("(", ",", ")") + "-> " + k.mkString(",")
  }
  commonFriendsLarge
    .saveAsTextFile("./output/results_friends_large.txt")


}

friends_data_small

C   -> P,M,U,H,R
F   -> Q,G,N,E,U
G   -> J,D,L,Z,T
M   -> X,Y,O,R,H
S   -> A,Q,E,L,D
T   -> Q,D,F,T,W
I   -> X,H,R,V,B
B   -> Q,F,K,Y,S
E   -> R,C,P,B,N
N   -> C,M,K,W,O

New_York_City_Leading_Causes_of_Death_2.csv

ScalaChallenge.scala

case class Accumulator(elements: List[(Int, Symbol)] = List.empty[(Int, Symbol)],
                       element: Option[Symbol] = None,
                       seqCount: Int = 0)

object ScalaChallenge01 extends App {
  def encode(decoded: List[Symbol]) = {

    if (decoded.nonEmpty) {
      val initial = Accumulator(List((1, decoded.last)), Some(decoded.last), 1)
      val encodings = decoded.init.foldRight(initial)((e, acc) => {
        acc.element match {
          case Some(element) if element == e => {
            val seqCount = acc.seqCount + 1
            val encodedElements = (seqCount, element) :: acc.elements.tail
            val nextElement = Some(element)
            Accumulator(encodedElements, nextElement, seqCount)
          }
          case Some(element) => Accumulator((1, e) :: acc.elements, Some(e), 1)
          case None => acc
        }
      })
      encodings.elements
    }
    else {
      List.empty[(Int, Symbol)]
    }
  }


    def decode[Symbol](encoded: List[(Int, Symbol)]): List[Symbol] = {
      @scala.annotation.tailrec
      def decodeRecurse[Symbol](encoded: List[(Int, Symbol)], acc: List[Symbol]): List[Symbol] = {
        encoded match {
          case List() => acc
          case head :: tail => decodeRecurse(tail, List.fill(head._1)(head._2) ::: acc)
        }
      }
      decodeRecurse(encoded, List.empty[Symbol]).reverse
    }




    val actualEncoding = encode(List('a, 'a, 'a, 'a, 'b, 'c, 'c, 'a, 'a, 'd, 'e, 'e, 'e, 'e))
    val expectedEncoding = List((4, 'a), (1, 'b), (2, 'c), (2, 'a), (1, 'd), (4, 'e))
    println("Question 1 Results:")
    println(s"Expected Encoding Result=${expectedEncoding}")
    println(s"Actual Encoding Result=${actualEncoding}\n")
    assert(actualEncoding == expectedEncoding)
    val actualDecoding = decode(List((4, 'a), (1, 'b), (2, 'c), (2, 'a), (1, 'd), (4, 'e)))
    val expectedDecoding = List('a, 'a, 'a, 'a, 'b, 'c, 'c, 'a, 'a, 'd, 'e, 'e, 'e, 'e)
    println("Question 2 Results:")
    println(s"Expected Decoding Result=${expectedDecoding}")
    println(s"Actual Decoding Result=${actualDecoding}")
    assert(actualDecoding == expectedDecoding)


  }