DAG_Generator_scala.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "distant-leisure",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[32mimport \u001b[39m\u001b[36m$ivy.$                                  ;\u001b[39m"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "// import Spark from Maven (dependency manager) repository\n",
    "import $ivy.`org.apache.spark::spark-sql:3.0.0`;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "voluntary-simpson",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[32mimport \u001b[39m\u001b[36morg.apache.log4j.{Level, Logger};\n",
       "\u001b[39m"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "// Disable logs to avoid clogging up cell output\n",
    "// RECOMMEND enabling for debugging purposes\n",
    "import org.apache.log4j.{Level, Logger};\n",
    "Logger.getLogger(\"org\").setLevel(Level.OFF);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "imposed-foundation",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "\u001b[32mimport \u001b[39m\u001b[36morg.apache.spark.sql._\n",
       "\n",
       "\u001b[39m\n",
       "\u001b[36mspark\u001b[39m: \u001b[32mSparkSession\u001b[39m = org.apache.spark.sql.SparkSession@548561bd\n",
       "\u001b[36msc\u001b[39m: \u001b[32morg\u001b[39m.\u001b[32mapache\u001b[39m.\u001b[32mspark\u001b[39m.\u001b[32mSparkContext\u001b[39m = org.apache.spark.SparkContext@4b47fc36"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.sql._\n",
    "\n",
    "val spark = SparkSession.\n",
    "            builder().\n",
    "            appName(\"scala-spark-notebook\").\n",
    "            master(\"spark://spark-master:7077\").\n",
    "            config(\"spark.executor.memory\", \"512m\").\n",
    "            .setJars(new String[]{\"/path/to/jar/with/your/class.jar\"}).\n",
    "            getOrCreate()\n",
    "\n",
    "val sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "cultural-salvation",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[32mimport \u001b[39m\u001b[36morg.apache.spark\n",
       "\n",
       "\u001b[39m\n",
       "\u001b[36mSEP\u001b[39m: \u001b[32mString\u001b[39m = \u001b[32m\",\"\u001b[39m\n",
       "\u001b[36mairportsData\u001b[39m: \u001b[32mspark\u001b[39m.\u001b[32mrdd\u001b[39m.\u001b[32mRDD\u001b[39m[\u001b[32mString\u001b[39m] = /data/airports_data2.csv MapPartitionsRDD[3] at textFile at cmd4.sc:13\n",
       "defined \u001b[32mfunction\u001b[39m \u001b[36mextractTimezoneToLatLong\u001b[39m\n",
       "\u001b[36mtzLatLong\u001b[39m: \u001b[32mspark\u001b[39m.\u001b[32mrdd\u001b[39m.\u001b[32mRDD\u001b[39m[(\u001b[32mString\u001b[39m, (\u001b[32mFloat\u001b[39m, \u001b[32mFloat\u001b[39m))] = MapPartitionsRDD[4] at map at cmd4.sc:22\n",
       "\u001b[36mdoubleLatLong\u001b[39m: (\u001b[32mString\u001b[39m, (\u001b[32mFloat\u001b[39m, \u001b[32mFloat\u001b[39m)) => (\u001b[32mString\u001b[39m, (\u001b[32mFloat\u001b[39m, \u001b[32mFloat\u001b[39m)) = ammonite.$sess.cmd4$Helper$$Lambda$3734/2065157289@3080dbab\n",
       "\u001b[36mtzLatLongDbl\u001b[39m: \u001b[32mspark\u001b[39m.\u001b[32mrdd\u001b[39m.\u001b[32mRDD\u001b[39m[(\u001b[32mString\u001b[39m, (\u001b[32mFloat\u001b[39m, \u001b[32mFloat\u001b[39m))] = MapPartitionsRDD[5] at map at cmd4.sc:26\n",
       "\u001b[36mtotalByTz\u001b[39m: \u001b[32mspark\u001b[39m.\u001b[32mrdd\u001b[39m.\u001b[32mRDD\u001b[39m[(\u001b[32mString\u001b[39m, (\u001b[32mFloat\u001b[39m, \u001b[32mFloat\u001b[39m))] = ShuffledRDD[6] at reduceByKey at cmd4.sc:29"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark\n",
    "\n",
    "val SEP = \",\"\n",
    "\n",
    "// Does not work for some reason..\n",
    "def removeHeader(rdd: spark.rdd.RDD[String]): spark.rdd.RDD[String] = {\n",
    "    val header = rdd.first()\n",
    "    return rdd.filter(line => line != header)\n",
    "}\n",
    "\n",
    "// Load data and drop header\n",
    "// val rawData = sc.textFile(\"/data/airports_data.csv\")\n",
    "\n",
    "// None work\n",
    "// val airportsData = rawData.filter(line  => !line.contains(\"airport\"))\n",
    "// val airportsData = removeHeader( rawData )\n",
    "// val airportsData = sc.textFile(\"/data/airports_data2.csv\")\n",
    "\n",
    "// Extract KV pair using function\n",
    "def extractTimezoneToLatLong(line: String): (String, (Float, Float)) = {\n",
    "    val nsplit = line.split(SEP)\n",
    "    return (nsplit(5), (nsplit(4).toFloat, nsplit(3).toFloat)) // Scala Lists are indexed as list(i)\n",
    "}\n",
    "val tzLatLong = airportsData.map(extractTimezoneToLatLong)\n",
    "\n",
    "// Double the Lat and Long using a lambda function\n",
    "val doubleLatLong = (tup: (String, (Float, Float))) => (tup._1, (2*tup._2._1, 2*tup._2._2))\n",
    "val tzLatLongDbl = tzLatLong.map(doubleLatLong) \n",
    "\n",
    "// Aggregate using inline lambda\n",
    "val totalByTz = tzLatLongDbl.reduceByKey(((v1, v2) => (v1._1 + v2._1, v1._2 + v2._2)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "funded-activation",
   "metadata": {},
   "outputs": [
    {
     "ename": "",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[31morg.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 5, 172.20.0.4, executor 1): java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.Aggregator.mergeCombiners of type scala.Function2 in instance of org.apache.spark.Aggregator\n\tat java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2301)\n\tat java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1431)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2411)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.readObject(ObjectInputStream.java:503)\n\tat java.io.ObjectInputStream.readObject(ObjectInputStream.java:461)\n\tat scala.collection.immutable.List$SerializationProxy.readObject(List.scala:488)\n\tat sun.reflect.GeneratedMethodAccessor2.invoke(Unknown Source)\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.lang.reflect.Method.invoke(Method.java:498)\n\tat java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1184)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2296)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2285)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.readArray(ObjectInputStream.java:2093)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1655)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.readArray(ObjectInputStream.java:2093)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1655)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.readObject(ObjectInputStream.java:503)\n\tat java.io.ObjectInputStream.readObject(ObjectInputStream.java:461)\n\tat scala.collection.immutable.List$SerializationProxy.readObject(List.scala:488)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.lang.reflect.Method.invoke(Method.java:498)\n\tat java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1184)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2296)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)\n\tat java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)\n\tat java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)\n\tat java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)\n\tat java.io.ObjectInputStream.readObject(ObjectInputStream.java:503)\n\tat java.io.ObjectInputStream.readObject(ObjectInputStream.java:461)\n\tat org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76)\n\tat org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:115)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:85)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:444)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:447)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\n\nDriver stacktrace:\u001b[39m\n  org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m2023\u001b[39m)\n  org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m1972\u001b[39m)\n  org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m1971\u001b[39m)\n  scala.collection.mutable.ResizableArray.foreach(\u001b[32mResizableArray.scala\u001b[39m:\u001b[32m62\u001b[39m)\n  scala.collection.mutable.ResizableArray.foreach$(\u001b[32mResizableArray.scala\u001b[39m:\u001b[32m55\u001b[39m)\n  scala.collection.mutable.ArrayBuffer.foreach(\u001b[32mArrayBuffer.scala\u001b[39m:\u001b[32m49\u001b[39m)\n  org.apache.spark.scheduler.DAGScheduler.abortStage(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m1971\u001b[39m)\n  org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m950\u001b[39m)\n  org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m950\u001b[39m)\n  scala.Option.foreach(\u001b[32mOption.scala\u001b[39m:\u001b[32m407\u001b[39m)\n  org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m950\u001b[39m)\n  org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m2203\u001b[39m)\n  org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m2152\u001b[39m)\n  org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m2141\u001b[39m)\n  org.apache.spark.util.EventLoop$$anon$1.run(\u001b[32mEventLoop.scala\u001b[39m:\u001b[32m49\u001b[39m)\n  org.apache.spark.scheduler.DAGScheduler.runJob(\u001b[32mDAGScheduler.scala\u001b[39m:\u001b[32m752\u001b[39m)\n  org.apache.spark.SparkContext.runJob(\u001b[32mSparkContext.scala\u001b[39m:\u001b[32m2093\u001b[39m)\n  org.apache.spark.SparkContext.runJob(\u001b[32mSparkContext.scala\u001b[39m:\u001b[32m2114\u001b[39m)\n  org.apache.spark.SparkContext.runJob(\u001b[32mSparkContext.scala\u001b[39m:\u001b[32m2133\u001b[39m)\n  org.apache.spark.SparkContext.runJob(\u001b[32mSparkContext.scala\u001b[39m:\u001b[32m2158\u001b[39m)\n  org.apache.spark.rdd.RDD.$anonfun$collect$1(\u001b[32mRDD.scala\u001b[39m:\u001b[32m1004\u001b[39m)\n  org.apache.spark.rdd.RDDOperationScope$.withScope(\u001b[32mRDDOperationScope.scala\u001b[39m:\u001b[32m151\u001b[39m)\n  org.apache.spark.rdd.RDDOperationScope$.withScope(\u001b[32mRDDOperationScope.scala\u001b[39m:\u001b[32m112\u001b[39m)\n  org.apache.spark.rdd.RDD.withScope(\u001b[32mRDD.scala\u001b[39m:\u001b[32m388\u001b[39m)\n  org.apache.spark.rdd.RDD.collect(\u001b[32mRDD.scala\u001b[39m:\u001b[32m1003\u001b[39m)\n  ammonite.$sess.cmd5$Helper.<init>(\u001b[32mcmd5.sc\u001b[39m:\u001b[32m1\u001b[39m)\n  ammonite.$sess.cmd5$.<init>(\u001b[32mcmd5.sc\u001b[39m:\u001b[32m7\u001b[39m)\n  ammonite.$sess.cmd5$.<clinit>(\u001b[32mcmd5.sc\u001b[39m:\u001b[32m-1\u001b[39m)\n\u001b[31mjava.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.Aggregator.mergeCombiners of type scala.Function2 in instance of org.apache.spark.Aggregator\u001b[39m\n  java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(\u001b[32mObjectStreamClass.java\u001b[39m:\u001b[32m2301\u001b[39m)\n  java.io.ObjectStreamClass.setObjFieldValues(\u001b[32mObjectStreamClass.java\u001b[39m:\u001b[32m1431\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2411\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.readObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m503\u001b[39m)\n  java.io.ObjectInputStream.readObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m461\u001b[39m)\n  scala.collection.immutable.List$SerializationProxy.readObject(\u001b[32mList.scala\u001b[39m:\u001b[32m488\u001b[39m)\n  sun.reflect.GeneratedMethodAccessor2.invoke(\u001b[32mUnknown Source\u001b[39m)\n  sun.reflect.DelegatingMethodAccessorImpl.invoke(\u001b[32mDelegatingMethodAccessorImpl.java\u001b[39m:\u001b[32m43\u001b[39m)\n  java.lang.reflect.Method.invoke(\u001b[32mMethod.java\u001b[39m:\u001b[32m498\u001b[39m)\n  java.io.ObjectStreamClass.invokeReadObject(\u001b[32mObjectStreamClass.java\u001b[39m:\u001b[32m1184\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2296\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2285\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.readArray(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2093\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1655\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.readArray(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2093\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1655\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.readObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m503\u001b[39m)\n  java.io.ObjectInputStream.readObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m461\u001b[39m)\n  scala.collection.immutable.List$SerializationProxy.readObject(\u001b[32mList.scala\u001b[39m:\u001b[32m488\u001b[39m)\n  sun.reflect.NativeMethodAccessorImpl.invoke0(\u001b[32mNative Method\u001b[39m)\n  sun.reflect.NativeMethodAccessorImpl.invoke(\u001b[32mNativeMethodAccessorImpl.java\u001b[39m:\u001b[32m62\u001b[39m)\n  sun.reflect.DelegatingMethodAccessorImpl.invoke(\u001b[32mDelegatingMethodAccessorImpl.java\u001b[39m:\u001b[32m43\u001b[39m)\n  java.lang.reflect.Method.invoke(\u001b[32mMethod.java\u001b[39m:\u001b[32m498\u001b[39m)\n  java.io.ObjectStreamClass.invokeReadObject(\u001b[32mObjectStreamClass.java\u001b[39m:\u001b[32m1184\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2296\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.defaultReadFields(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2405\u001b[39m)\n  java.io.ObjectInputStream.readSerialData(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2329\u001b[39m)\n  java.io.ObjectInputStream.readOrdinaryObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m2187\u001b[39m)\n  java.io.ObjectInputStream.readObject0(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m1667\u001b[39m)\n  java.io.ObjectInputStream.readObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m503\u001b[39m)\n  java.io.ObjectInputStream.readObject(\u001b[32mObjectInputStream.java\u001b[39m:\u001b[32m461\u001b[39m)\n  org.apache.spark.serializer.JavaDeserializationStream.readObject(\u001b[32mJavaSerializer.scala\u001b[39m:\u001b[32m76\u001b[39m)\n  org.apache.spark.serializer.JavaSerializerInstance.deserialize(\u001b[32mJavaSerializer.scala\u001b[39m:\u001b[32m115\u001b[39m)\n  org.apache.spark.scheduler.ShuffleMapTask.runTask(\u001b[32mShuffleMapTask.scala\u001b[39m:\u001b[32m85\u001b[39m)\n  org.apache.spark.scheduler.ShuffleMapTask.runTask(\u001b[32mShuffleMapTask.scala\u001b[39m:\u001b[32m52\u001b[39m)\n  org.apache.spark.scheduler.Task.run(\u001b[32mTask.scala\u001b[39m:\u001b[32m127\u001b[39m)\n  org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(\u001b[32mExecutor.scala\u001b[39m:\u001b[32m444\u001b[39m)\n  org.apache.spark.util.Utils$.tryWithSafeFinally(\u001b[32mUtils.scala\u001b[39m:\u001b[32m1377\u001b[39m)\n  org.apache.spark.executor.Executor$TaskRunner.run(\u001b[32mExecutor.scala\u001b[39m:\u001b[32m447\u001b[39m)\n  java.util.concurrent.ThreadPoolExecutor.runWorker(\u001b[32mThreadPoolExecutor.java\u001b[39m:\u001b[32m1149\u001b[39m)\n  java.util.concurrent.ThreadPoolExecutor$Worker.run(\u001b[32mThreadPoolExecutor.java\u001b[39m:\u001b[32m624\u001b[39m)\n  java.lang.Thread.run(\u001b[32mThread.java\u001b[39m:\u001b[32m748\u001b[39m)"
     ]
    }
   ],
   "source": [
    "totalByTz.collect()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Scala 2.12.10",
   "language": "scala",
   "name": "scala"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".sc",
   "mimetype": "text/x-scala",
   "name": "scala",
   "nbconvert_exporter": "script",
   "version": "2.12.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}