From 190f2db9c81868b295f5a5b559ab6fb502452a1b Mon Sep 17 00:00:00 2001 From: enricogallinucci Date: Wed, 3 Apr 2024 22:26:59 +0200 Subject: [PATCH] update first class --- materials/2024_bbs_dm_spark_basics.ipynb | 877 +---------------------- 1 file changed, 1 insertion(+), 876 deletions(-) diff --git a/materials/2024_bbs_dm_spark_basics.ipynb b/materials/2024_bbs_dm_spark_basics.ipynb index 8059695..08e8882 100644 --- a/materials/2024_bbs_dm_spark_basics.ipynb +++ b/materials/2024_bbs_dm_spark_basics.ipynb @@ -1,876 +1 @@ -{ - "nbformat": 4, - "nbformat_minor": 0, - "metadata": { - "colab": { - "provenance": [], - "collapsed_sections": [ - "oBd7XwkFBDEF" - ] - }, - "kernelspec": { - "name": "python3", - "display_name": "Python 3" - }, - "language_info": { - "name": "python" - } - }, - "cells": [ - { - "cell_type": "markdown", - "source": [ - "# Install Spark & initialize application\n", - "\n", - "Run the following code to install Spark in your Colab environment." - ], - "metadata": { - "id": "EsElqAaj4Sse" - } - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": { - "id": "fbvEUbWIHm2s" - }, - "outputs": [], - "source": [ - "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n", - "!wget -q https://archive.apache.org/dist/spark/spark-3.5.1/spark-3.5.1-bin-hadoop3.tgz\n", - "# If the above command is too slow, uncomment the following and try it\n", - "# !wget -q https://big.csr.unibo.it/downloads/bbs-dm/spark-3.2.1-bin-hadoop2.7.tgz\n", - "!tar xf spark-3.5.1-bin-hadoop3.tgz\n", - "!pip install -q findspark" - ] - }, - { - "cell_type": "code", - "source": [ - "import os\n", - "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n", - "os.environ[\"SPARK_HOME\"] = \"/content/spark-3.5.1-bin-hadoop3\"\n", - "import findspark\n", - "findspark.init()\n", - "findspark.find() # Should return '/content/spark-3.5.1-bin-hadoop3'" - ], - "metadata": { - "id": "4oTFM5YtJvv7" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "from pyspark.sql import SparkSession\n", - "\n", - "spark = SparkSession.builder\\\n", - " .master(\"local\")\\\n", - " .appName(\"Colab\")\\\n", - " .config('spark.ui.port', '4050')\\\n", - " .getOrCreate()\n", - "sc = spark.sparkContext\n", - "\n", - "sc" - ], - "metadata": { - "id": "KJlzVAmbJ9vL" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "# Spark: working with RDDs\n", - "\n", - "Check the documentation: [here](https://spark.apache.org/docs/latest/api/python/reference/pyspark.html#rdd-apis)." - ], - "metadata": { - "id": "oBd7XwkFBDEF" - } - }, - { - "cell_type": "markdown", - "source": [ - "## Basics" - ], - "metadata": { - "id": "jcoFwGvm4gj6" - } - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": { - "autoscroll": "auto", - "id": "vlFswJyWytKG" - }, - "outputs": [], - "source": [ - "# let's create a simple example\n", - "riddle1 = \"over the bench the sheep lives under the bench the sheep dies\"\n", - "riddle2 = [\"over the bench the sheep lives\", \"under the bench the sheep dies\"]" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": { - "autoscroll": "auto", - "id": "WHywj4BmytKH" - }, - "outputs": [], - "source": [ - "# create an RDD from the `riddle` string\n", - "rdd1 = sc.parallelize(riddle1.split(\" \"))\n", - "# each tuple of the RDD corresponds to a single word\n", - "\n", - "print(rdd1)\n", - "# why is there no result returned?" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": { - "autoscroll": "auto", - "id": "3oYjBLnOytKI" - }, - "outputs": [], - "source": [ - "# compute the RDD\n", - "print(rdd1.collect())" - ] - }, - { - "cell_type": "code", - "source": [ - "rdd2 = sc.parallelize(riddle2)\n", - "print(rdd2.collect())" - ], - "metadata": { - "id": "exzQLruZ9qgg" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "## Transformations" - ], - "metadata": { - "id": "iyH7halU9HiB" - } - }, - { - "cell_type": "code", - "source": [ - "# map: returns a new RDD by applying a function to each of the elements in the original RDD\n", - "rdd1.map(lambda s: s.upper()).collect()" - ], - "metadata": { - "id": "sndBHyEF86T5" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# flatMap: returns a new RDD by applying the function to every element of the parent RDD and then flattening the result\n", - "rdd2.flatMap(lambda s: s.split(\" \")).collect()" - ], - "metadata": { - "id": "9MlPRBd_-Cl1" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# filter: returns a new RDD containing only the elements in the parent RDD that satisfy the function inside filter\n", - "rdd1.filter(lambda s: s.startswith(\"u\")).collect()" - ], - "metadata": { - "id": "UlT_jxmH9Myx" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# distinct: returns a new RDD that contains only the distinct elements in the parent RDD\n", - "rdd1.distinct().collect()" - ], - "metadata": { - "id": "QxxJdRxW-Xcj" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# groupByKey: groups the values for each key in the (key, value) pairs of the RDD into a single sequence\n", - "rdd1.map(lambda s: (s,1)).groupByKey().mapValues(list).collect()\n", - "\n", - "# (first map converts to a key-value RDD)\n", - "# (mapValues is a map that operates only on the values - in this case, used to convert from ResultIterable to List for printing reasons)" - ], - "metadata": { - "id": "dBAh2Gs8-fdM" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# reduceByKey: when called on a key-value RDD, returns a new dataset in which the values for each of its key are aggregated\n", - "rdd1.map(lambda s: (s,1)).reduceByKey(lambda x, y: x + y).collect()" - ], - "metadata": { - "id": "cH1dxiGl_cS5" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# sortByKey: returns a new RDD with (key,value) pairs of parent RDD in sorted order according to the key\n", - "rdd1.map(lambda s: (s,1)).sortByKey().collect()" - ], - "metadata": { - "id": "-XJWbbv6_0Tw" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# join: starting from two RDD with (key, value1) and (key, value2) pairs, returns a new RDD with (key, (value1, value2)) pairs\n", - "rddA = sc.parallelize([(1, \"A1\"), (2, \"A2\"), (3, \"A3\")])\n", - "rddB = sc.parallelize([(1, \"B1\"), (2, \"B2\"), (4, \"B4\")])\n", - "rddA.join(rddB).collect()" - ], - "metadata": { - "id": "SGIn90U0Md8j" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "## Actions" - ], - "metadata": { - "id": "gKJ_7MCsAIy_" - } - }, - { - "cell_type": "code", - "source": [ - "# collect: returns a list that contains all the elements of the RDD\n", - "rdd1.collect()" - ], - "metadata": { - "id": "XQXX-d20_vWo" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# count: returns the number of elements in the RDD\n", - "rdd1.count()" - ], - "metadata": { - "id": "FXuL9K2qATfL" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# reduce: aggregates the elements of the RDD using a function that takes two elements of the RDD as input and gives the result\n", - "sc.parallelize([1, 2, 3, 4, 5]).reduce( lambda x, y: x * y)" - ], - "metadata": { - "id": "YI10eDDnAWe7" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# take: returns the first n elements of RDD in the same order\n", - "rdd1.take(2)" - ], - "metadata": { - "id": "CgN5qtwBAvIv" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# saveAsTextFile: saves the content of the RDD to a file\n", - "rdd1.saveAsTextFile(\"rdd1\")" - ], - "metadata": { - "id": "yS8SuYRQvV6F" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "## Examples" - ], - "metadata": { - "id": "gmosJNIOA5S_" - } - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": { - "autoscroll": "auto", - "id": "mGRkO7_oytKK" - }, - "outputs": [], - "source": [ - "# Flatten the words beginning with the letter C\n", - "# - map: transform each string in upper case (remember: map returns a new RDD with the same cardinality)\n", - "# - filter: keep only the strings beginning with \"C\" (remember: filter returns a new RDD with the same or smaller cardinality)\n", - "# - flatMap: explode each string into its characters (remember: flatMap returns a new RDD with the any cardinality)\n", - "rdd1\\\n", - " .map(lambda s: s.upper())\\\n", - " .filter(lambda s: s.startswith(\"U\"))\\\n", - " .flatMap(lambda s: list(s))\\\n", - " .collect()" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": { - "autoscroll": "auto", - "id": "Wrr37cZ2ytKL" - }, - "outputs": [], - "source": [ - "# A simple word count\n", - "# - map: map each word to a tuple (word, 1); each tuple represent the count associate with a word\n", - "# - reduceByKey: group all the tuples with the same word and sum the counts\n", - "# - sortBy: sort tuples by count\n", - "rdd1\\\n", - " .map(lambda s: (s, 1))\\\n", - " .reduceByKey(lambda a, b: a + b)\\\n", - " .sortBy(lambda x: x[1], False)\\\n", - " .collect()" - ] - }, - { - "cell_type": "code", - "source": [ - "# Compute average length of words depending on their initial letter\n", - "# map: map each word to a key-value tuple (word, (wordLength, 1)), where the value is an object composed by two value: the length of the word and a 1\n", - "# reduceByKey: group all the tuples with the same word to 1) sum the lengths, and 2) sum the counts\n", - "# mapValues: divides the sums by the counts to compute the averages\n", - "# sortBy: sort tuples by averages\n", - "rdd1\\\n", - " .map(lambda s: (s[0], (len(s),1)))\\\n", - " .reduceByKey(lambda a, b: (a[0] + b[0], a[1] + b[1]))\\\n", - " .mapValues(lambda x: x[0]/x[1])\\\n", - " .sortBy(lambda x: x[1], False)\\\n", - " .collect()" - ], - "metadata": { - "id": "zxBoMsnTCATh" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "# Spark: working with DataFrames\n", - "\n", - "Check the documentation: [here](https://spark.apache.org/docs/3.2.1/api/python/reference/api/pyspark.sql.DataFrame.html).\n", - "\n", - "What is different from Pandas' DataFrames?\n", - "\n", - "- Spark supports parallelization (Pandas doesn't), thus it's more suitable for big data processing\n", - "- Spark follows Lazy Execution, which means that a task is not executed until an action is performed (Pandas follows Eager Execution, which means task is executed immediately)\n", - "- Spark has immutability (Pandas has mutability)\n", - "- The data structure is similar, the APIs are different" - ], - "metadata": { - "id": "sCXrRXAAEtet" - } - }, - { - "cell_type": "code", - "source": [ - "# !wget https://raw.githubusercontent.com/w4bo/2023-bbs-dm/master/materials/datasets/housing.csv\n", - "!wget https://big.csr.unibo.it/downloads/bbs-dm/datasets/housing.csv\n", - "df = spark.read.option(\"delimiter\", \",\").option(\"header\", \"true\").csv(\"housing.csv\")\n", - "df.show()" - ], - "metadata": { - "id": "IGChYOQWRZku" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# Switching from Spark to Pandas\n", - "pandasDF = df.toPandas()\n", - "print(pandasDF)" - ], - "metadata": { - "id": "_aSykf1wV8dB" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# Switching from Pandas to Spark\n", - "df = spark.createDataFrame(pandasDF)\n", - "df.show()" - ], - "metadata": { - "id": "cFUtMUaUWTY5" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# select: returns a new DataFrame with only selected columns (similar to a map on RDDs)\n", - "df.select('population','median_house_value').show()" - ], - "metadata": { - "id": "eFAs2zBHPLTA" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# select, similarly to a map, allows column values to be redefined\n", - "df.select(df.population,df.median_house_value/1000).show()\n", - "# put the operation within parenthesis and add .alias('median_house_value_in_K$')" - ], - "metadata": { - "id": "FmClDtFaQqO7" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# withColumn: used to manipulate (rename, change the value, convert the datatype)\n", - "# an existing column in a dataframe (or to create a new column) while keeping the rest intact\n", - "df.withColumn('median_house_value_in_K$',df.median_house_value/1000).show()" - ], - "metadata": { - "id": "4whu8BylWfAK" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# filter: returns a new DataFrame containing only the elements in the parent DataFrame that satisfy the function inside filter (as in RDDs)\n", - "# orderBY: orders the DataFrame by the selected column(s)\n", - "df.filter(df.population > 1000).orderBy(df.population.asc()).show()" - ], - "metadata": { - "id": "9zrnJn3_Q0PD" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# groupBy: returns a new DataFrame which is the result of an aggregation\n", - "df.groupBy(df.ocean_proximity).agg({'median_house_value': 'avg', '*': 'count'}).show()" - ], - "metadata": { - "id": "1qv5_B90Raho" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# withColumnRenamed: rename a column\n", - "df.groupBy(df.ocean_proximity).agg({'*': 'count'}).withColumnRenamed(\"count(1)\", \"tot\").show()" - ], - "metadata": { - "id": "E3b1rxzMN7Lm" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "# SQL queries can be run on a DataFrame\n", - "df.createOrReplaceTempView(\"housing\")\n", - "spark.sql(\"select ocean_proximity, avg(median_house_value) as avg_price from housing group by ocean_proximity order by avg_price desc\").show()" - ], - "metadata": { - "id": "TaP-GatdTD7k" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "# Exercise: creating a cube\n", - "\n", - "You are working with two files:\n", - "\n", - "- weather-stations.csv: it contains a list of weather stations that capture weather information every day of every year throughout the world\n", - " - Each station is identified by a StationID\n", - "- weather-sample-10k.csv: it contains the data measured by a certain station on a certain date (a sample of 10k lines collected from the National Climatic Data Center of the USA)\n", - " - Each weather measurenent is identified by a StationID and a Timestamp\n", - "\n", - "Your goal is to create a single file representing the following cube and to run some queries through PowerBI.\n", - "\n", - "![image.png](data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAPwAAAEICAYAAAB735ncAAAgAElEQVR4nO3deVyU5f7/8RcyICoiASoomwqklhuKo6HHSu1Y5pJQtp1jZd+Kk5Sl/fppx6I62befWZkZns7Bkx23TmhpR1sIWxQR9zRRVglQkU1kUZYb5vfHzMC4A94sM/fn+Xj4eDgz133dH+6Z91zXfc99z9gpimJACKEJHdq6ACFE65HAC6EhEnghNEQCL4SGSODbrXiedwzl4zTDJffNIU4Os4pmksC3W+OJXN6BtXHp1Oc7fgv7P3ieCXZtWZewZhL4dizg6ShGzF3GDwaAdP4etY+HJwYgeRfNJYFv18YzNeIToqLTMMQvJ3JIFBGBdhin9jp0Oh06ncW0P/55HHVXuX9ONNFjHdCFfkya7BJolgS+nRsfuZwOa5fx3Gv7+OD5CdgRz/OOX3JvlYKiKCjH/8z6x6KNIR6/jGrFdP93I5i77IeG3YGVazCsqkZJ+AuBMkXQLF1bFyCuI+BpokY48sfardQE2kF6OgcMK1nhsNKi0TNkAYGk8/exA3g20RRz/c2kGyYSCPCMeXYgtExGeCvgN1CPfqB/wx0jP+RYjWkkVxQU5SMm2hnDfuTVatPIv4LRbVeyaKck8NYmIIDgvc+x7IdLd8R/JzlpJOb3hfT4Nexu9eJEeydTeqsznmXJKxg7wAGdOfPPfEPNRxOJXB7FgAEOPAfon4lAf52eNu/NZldKPr/lnCXlVCn+3Z3RB3ZnZIAHk4O90dnLeGBr7ORqOe3Jyi/nyZUJKHV1TAvxY6i/GwGe3cgtqmBfRgHxR06TnlfK6jljGd7Xva3LFSqSwGvM2h0ZzPtsLy9PH8z8qYOu2m7NL+m89NkeFtw3mGcnDWjFCkVLksBryPGTJYxZtI3ExVPo39v1uu3zSi4wbP6XbH55goz0NkJ20jRCqa3jyZW7eO3+YY0KO4Cnayeinwpl1kc7qKypbeEKRWuQwGvE1gO5ODnYM/feW5u03PSRfowK6kFMfGoLVSZakwReI/akFzJ+UK9mLTtuoCd70gtVrki0BQm8RiSknGFoH7dmLTu0jxtJaQUqVyTaggReI7Lyyxnq37wDb0P93UnPK1W5ItEWJPAa4d/DmeMnS5q1bHpeKd7uXVSuSLQFCbxG6AO6cyiruFnLHjpRhD6wu8oVibYggdeI4L7u/Hw0r1nL/pycx8gAD5UrEm1BAq8RYXo/sgrK2JCQ2aTldh47Q2ziCWaNC2ihykRrksBrhM6+A6vnjCUyJpHcoopGLVNeWcPjK35h2eN6PFycWrhC0Rok8BoyxM+NBdMHMXbRVmITT1yz7U9HTzPoxS+ZFuJL2Cj/a7YV1kPOpdeghOP5zFqxgzH9ezItxJehfdwJ8HQxXS1XSPyRU/x3fw7LHh/J5GCfti5XqEgCr1HllTWs3ZHBL8lnOJRVTMqpc/Ry60xIPw9GBnjw9MT+uHZxbOsyhcok8IIL1Qo/H88GYNLgvm1cjWhJsg8vhIZI4IXQEAm8EBoigRdCQyTwQmiIBF4IDZHAC6EhEnghNEQCL4SGSOCF0BAJvBAaIoEXQkMk8EJoiAReCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggNkcALoSESeCE05OqBrzrMkuE6dDrjP69Xd7diWTYqez33uer401elV3hwD1HeV3us+fZEeavep7BeVwx81eElDO/yR8pWKCiK8V/aqIOs/b1liijfPEs7byi36Mn873YujWD2+reJ89PT8Ub73xOF95++uqx/IeCKgS/n26XvMXnnad4Y1XCv8z0RPOLXeoXZrgEEZy8h+tdKi/uySfiPO/c/2qPNqhLacHngy+PZ9P3/cM+oK7Q2N9k8q36qr3OZWj/yXzpSW942/j+OzbNMy3m9ym7T/a5haylYPMbUVxWHl4QyNXoZs3Q6XEJDGW5qC5h2Nbyw3glBIJMe6sGWb1Ooj/yeVbzkfC+3nNlLlUXLPVHeDds5+B3M7xHZ6+8j9J2t/Os+14sf2xOF921/I299OG66YN6pf1NJIsrb1I+M/pp2WeCrMlNJD7kZr6ssUL55Fq7P+LHTNNWv2DmW90ZZBPIaChb/mf0RCopSwsa7PmDx2t9xnraako2P0H3hTpTSLaZZRCXb3igiQlEoTfiOV+/6B9tMK6hK+Y6ver/L7Gu8IbV33g8tYOLyd/nWlLw923YROX8S3Sza7Iny5raMjyg2befMl3Yz7n8awnr0lUXkR+WhKJn82+9t3v22FEZGkbvrr3g+FEuxcoCXhzgBsD78M4YmKyjKLv764xuXzC6EllwW+I59gwjYm8LpKzavIjP1FAu/fANz3joOjrgokNfSfeGXpt0EZ8bPuI+9KVdeCzhxz7uzTetwZvyMu/jHtt1AFSnffYX7zDFY997FSO558kc2bi+F7PW8HTeeSTc7WTyeTcavk4hdMR0X0z2+Dy1g7o8b2W5K/C1v/YvnhzgBvoQ+cAeZKSe4Wowfil3BdBfzevP57UR1S/1hop27fErv0IdbezUuwK3FedI8Xkzcxu6qFL77fjwLbeBgwsgnllC+JJqt8f/B/aUIhjhdfxkhbtTlge84mIhX72LxXVMtjspXcXjZS6z9vSN9g3qx+L6GKXzV4WjeMO3zO/S5lV7/2GZ6rJz4TWvVqbLjzfzxrkQ2rvqO70ffgxXP5hv4hvJAj7eZ9j/O3Huny6UP0m/It4Q/2zCFz17/Nh/cEcZlTYVoAt2V7nSetpqKnUu4rZ+OWQB05p7VR9jiB/h9woHU2wjWLTY27nwPq49sMYZwcASv3uXKGN1ioDsREY80qgjn8TO4K2wMug/uYfWRLxh0WYuO3PzHu9g+ZgcvHnmp/t7Ne7PZlZLPbzlnSTlVin93Z/SB3RkZ4MHkYG909u35vCJfHlowl3+Nuo9JVwjxyKhd/Pu+wbjpyo13DH6L/buMU/ySa3U78h6e/PE23HSDeWv/Lu64RlPz9jucfZajOWfxdHVix9ESK9l+ojnsFEUxtHURjVF1eAl3/rUX67Y8giG/nCdXJqDU1TEtxI+h/m4EeHYjt6iCfRkFxB85TXpeKavnjGV4X/e2Lr3dyZLtp1lWEvhyNs/qRcxdR5jpqzDvs728PH0w86dePhcwW/NLOi99tocF9w3m2UkDWrHW9m3tjgzZfhrW7gNfdXgJtwUv4NTCnfz4dH/GLNpG4uIp9O/tet1l80ouMGz+l2x+eYKMVMDxkyWy/TSu3QfeTKmt4/aob5l5Wx/m3ntro5f7as/vLFi7j33vTMHJwb4FK2zfZPsJsKKr5bYeyMXJwb5JL1aA6SP9GBXUg5j41BaqzDrI9hNgRYHfk17I+EG9mrXsuIGe7EkvVLki6yLbT4AVBT4h5QxD+7g1a9mhfdxISitQuSLrIttPgBUFPiu/nKH+zTtwNNTfnfQ8bV8yIttPgBUF3r+HM8dPXvOUk6tKzyvF272LyhVZF9l+Aqwo8PqA7hzKKm7WsodOFKEP7K5yRdZFtp8AKwp8cF93fj6a16xlf07OY2SAh8oVWRfZfgKsKPBhej+yCsrYkJDZpOV2HjtDbOIJZo0LaKHKrINsPwFWFHidfQdWzxlLZEwiuUUVjVqmvLKGx1f8wrLH9Xi4aPv6U9l+Aqwo8ABD/NxYMH0QYxdtJTbxxDXb/nT0NINe/JJpIb6EjfJvpQrbN9l+wmpOrbWUcDyfWSt2MKZ/T6aF+DK0jzsBni6mq70KiT9yiv/uz2HZ4yOZHOzT1uW2O7L9tMsqAw/G6ebaHRn8knyGQ1nFpJw6Ry+3zoT082BkgAdPT+yPaxfHti6z3ZLtp01WG3hLF6oVfj6eDcCkwX3buBrrI9tPO6xqH14IcWMk8EJoiAReCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggNkcALoSESeCE0RAIvhIZI4IXQEAm8EBoigRdCQyTwQmiIBF4IDVE18GdjH0Sn0xn/9XuL/Wp2LlrW2Z18PCmo4fnT6eg37yea9+NUor1SLfDn97/F7S8PIklRUBSF0v+48dt+gLPEPtiPt66b/vPsfyuUmbG5apUkmqrnY/yQb3z+FCWPlVWz6DE+ht+vs1hh3BxC5c3BKqgW+KoTRyh/YhLDTbc7D49g1vBrLiLaNQ8mfrSfbwas4sOfJMq2QrXA3zQ+nJDX7rpkhD5L7IPdeTD2d17T63CfGUvu+f28NcRi2vjWfoyj+2j0ryWx8UH/ht2Bs7E8aJ5ius9EBv/W5sGwacPY/fVhiikkbo5fw5TfNPIXxs1h+N0rSVo2gR668cT8DlzSdnzM9eYIorWotw9/UzgbSr9n8Jv+6HTmKfxNhG8oYEO4H68nKRR9Ho535+G88qtp2liaxBPb/h+xuZ0Z/koiSa/rCduQhZLxCsPPxvLgiBReMu8ifD+Y9+fFIplvKx5M/Oj3+um+eeT3mPgR+795Bv3zP5CvxDPbr5C4OdP4Nvygqe1xwlc9gWS+fVD3KL05zAXvcER/9f32+oN7LnpeS9pHypnL25zPTOHY76+hN40SLvrXSNqXwhWaihbmM7APbgDp0YTqdOh0nty9MolDGWVXaF1C+sEklk3oYRrh+xOZlEXyCdktaA9a5mO5m8KJ3hDCqm8vT/zZ2AcJ+CKcLEVBUQrYEH71bpzDNpjamf5lvIIcFmhN6XwR9Q29+3U1hv2P6bxlOqh3fLn+GsuN4+8ZFs+bksHS291arWpxdaoF/mxspMWIfp7MlGNXaGW833VwH7wBzsYTG3vl/jr3vRmfjS+zWj7bayPpRIf2591RMbxyuxuFJ46S6z+QPsahnrg1SVdZzpWAYeks/lCO2rdHOrU6uumex2C0Dt1R4+1uYRv49XPjeDw+PIQH9TreC9vArysXEdJdj+41wC+CiPoRvjMDJt3Dcb0/Or/XScp4hVVJKYzW63jN1MLv9SQyXmkY4zfvzWZXSj6Hs89yNOcsnq5O7DhawsgADyYHe6Ozl/OKruWi7ZeWRk3Wp0zo8Wn94+P+nkHGbD/jjYlz+b9R/emnexoYR0REwwjvMWwaw56cQI9l4/h7RjyzP4pnfmh/ephfXX7P88Pepcgg3/as8scks/LLeXJlAkpdHdNC/Bjq70aAZzfTzx0XEH/kNOl5payeM5bhfd3butx2R7afdlld4NfuyGDeZ3t5efpg5k8ddNV2a35J56XP9rDgvsE8O2lAK1bYvsn20zarCvzxkyWMWbSNxMVT6N/b9brt80ouMGz+l2x+eYKMVMj2E1Z08YxSW8eTK3fx2v3DGvViBfB07UT0U6HM+mgHlTW1LVxh+ybbT4AVBX7rgVycHOyZe++tTVpu+kg/RgX1ICY+tYUqsw6y/QRYUeD3pBcyflCvZi07bqAne9ILVa7Iusj2E2BFgU9IOcPQPs37XGdoHzeS0gpUrsi6yPYTYEWBz8ovZ6h/8w4cDfV3Jz2vVOWKrItsPwFWFHj/Hs4cP1nSrGXT80rxdu+ickXWRbafACsKvD6gO4eymney5qETRegDu6tckXWR7SfAigIf3Nedn4/mNWvZn5PzGBngoXJF1kW2nwArCnyY3o+sgjI2JGQ2abmdx84Qm3iCWeMCWqgy6yDbT4AVBV5n34HVc8YSGZNIblFFo5Ypr6zh8RW/sOxxPR4uTi1cYfsm26850ogO1V30xZ6RcQBxROoiiWuFdUeqvBKrCTzAED83FkwfxNhFW4lNPHHNtj8dPc2gF79kWogvYaP8W6nC9k22X3PoWX6s4dr+5RNbbk1p0aGERqeZbgUSkaD++lS7PLa1PD/5Fkb0686sFTvYvDebaSG+DO3jToCni+lqr0Lij5ziv/tzWPb4SCYH+7R1ye2KbD9ts6oR3iy0fw8OLZnK6CAPYhNPMOXtOOzCY9Av2MKH247g5erEvv+dIi/Wq7Dcfut3ZjDxjW+wC48h9K//5dMfU2X7NVWa+au/dOhMU/206FB0lvPxtGhCQ6NJu3Q3wdQmLTqUAZFJJEUOMPWRRnRoKPUD/kXrsJzqm9rFNTzeMEu4nNWN8GbOTg48PbE/T0/sD0BucRm/5RbQpaMDY2+WF+r1mLffjFG+7D9hPHo/oo8nHl07t3Fl7U0SkQN0RALG6X0CEYEWD6dFE/oYfKooBALERaKLjENZHkWEbgtxyycyEUj7YQ08+imBBBKYoBBhXJjo0MeITptIREQCxwjlMT4lISIQSCPdch0D1vDoMYUE40qI1IUSXV9LEpFRj3JMUQhMiyZ0wIfERSznSnsDVjnCX4mDzvin1NTWtXEl1kWx2F4O9vZtWEl7ZbkPf0nYAbKSSUqKZIB59L07Gg6kk8ZEpkZEsyUOII0f1gQTZV44LtI0Ug8gMimJ5KzrlJCVDMs/tVj3RJ5bDmt+MI/kepZ/GmF8wwmcwKP6A6RfZZC32hH+Uo6mF2uNUktNbR0O8vVWjWL5Bml+0xRNFPENyhWOrgVOjeDuLXEsZwuRwVNRwBj2qIHG0RiIi9SxpRVLtZln2BxwA1BTK9duN5Z5hLdDRvhm8R+IPjqKK+42T3yO5Qe2ELnlAMufM74hpKUfgOAA42hMHFuiG7cOIh+zWEccH0bCoxMunW5cn+0EXtfwYq1RZFrfWOYR3kFnL7OiKzLuw+uudkAsMIKEb4Kv0iaQCY8eIPrAo5izGRgRRUT03aa2WzDtzBsfm/Ao1B+0u9Y67oZvrrB70QhW9RVX12IwGIhP/h2ltk4OPjXBb7kF5BaXycFOjbCZt3Q7O7v6EUoO3DWeeUov03ltsJnAQ8OBu2qZ0jdaw5Tepl4K4ips6llu+GhODto1lnmEd5QRXhNsK/Dmj+ZkSt9oNXXmKb1NvRTEVdjUs+xoLyN8UykypdcUmznxBho+mpOP5Rqntq6uYR9eY1N6g8FAful5enbrwi/HczhfXdPWJbWIzo4O/KF/w6cvthV40whfLSN8o9TU1mEwGD+V1dqUPqvgHCl5xfTspq3v6rOtwOtkH74pLj6tVjsj/JlzFWQWGL/Q07WzE8P8erZxRa3Hpt7W6/fhFRnhG8PywhlHjYzwlTUKOcVl1NTW4dmtCz5uXdu6pFZlU8+y5VF6OXB3fTUavFIup7iMwrLzdHLU4ePugk4jb3RmNvXXWh5plpNvrq/+whmLsxRtWX7peXKLjD+o4ePmgrtzpzauqPXZ1LPsaG+PnZ0dIPvxjWGeBTnYd7D5ka5KqSW3uJQqpZaeLl3w1thU3symnmWdfQeL8+llSn89WjrLLreolPzS83R00OHt3hVHDR2ktGRTgQfLL8KQEf56tHIefWHZeXKKywDwcetKdw1fSWlzz7ScT994Wjittqa2lpziMiprFLp37azZqbyZzT3T5qPN1bIPf11auDQ2p6iMM+cqcNTZ4+3WFScHmzr1pMlsLvCOOvksvrFsfUpfVH6BXNNU3tutq+bOqrsSm3um5Yq5xrPlg3ZKbR25xWWcr67B3bkTPm4ubV1Su2CDgZd9+May/FjO1uQWl3G6pBydfQd83Fzo5KjtqbyZzT3TjnLFXKMpFl9gaUvOVlSSU2w+waYrnq4ylTezucDLFXONYzz92PaO0tfVGcgpLqWiqga3Lk54y1T+IrbzTJt06egAQG2dTXwZb4vRdbCr/7/57ERbkFNcyqmz5dh3sMPbzaX+9SCMWnTHJreogo27s/j+8Cky8sooLKukpKK6JVcp2inXLo64dulIv55d+cOAnswM7UOAp7qjb8n5Souj8i70uslZ1f5tQYt8L71SW8eybcks2fIb00J8mRbiS6BXNzxdO+PaxVHt1QkrUFJRTWFZJWmnzvHdryf5PCGTR8b24/WZw3ByuPFjCAYDHD1p/I59185O3OLtQVcnea1dSvXAHz9ZwuzoBDo56lj1l7H06antM5vElRWWVvLMJwkczSkhJiIUfWD3G+ovu6iU5JOF2NnBLb27a/6MuqtRNfDHT5YwZtE2Fs4YwkvTBqvVrbBhX+zK5LlVu1n7/DjGDfRsVh/nLlRxNLeQ0gtV+Li7cEtvD5WrtB2q7cMrtXXMjk6QsIsmuf+2vjg52jM7OoFDS6bi7NT0g2y5xWWUXqjCpVNHzX2DTVOpdpR+2bZkOjnqJOyiyaaM8GPcQE9eWX+gycvmFpeRU9TwmbtLp45ql2dTVAl8blEFS7b8xqq/jFWjO6FB7z+mZ/PebPZnFjV6mbLK6voTbLzduspn7o2gSuA37s5iWoivHKATzXaTc0eemnAz63ZmNnqZnKJSzp2voquTI95uXbGh0wlajCqB//7wKaaF+KrRldCw6SP9+O7QyUa1PXm27KIr4Vw7O7VkaTZDlcBn5JUR6NVNja7alqGEz6NicHp9D/nm+2pSiHrq4vsMBUmEhf+L2d9X3Pg6lSw+eD4G5w9/u/G+rFxgr26k55Vet115VQ25xWXUGQz0uslZpvJNoErgC8sq8XS1ga8NsuvKoL7OVKcVc8z0y0O1qXlsLYaq40UcMd2npJTwk86NSYObcVFGXT7fffgFAR9LwC/l5GB/0XflX01uUSlnKyrp0tEBbzcX7DvIXL6xVAl8SUW1jZxBZ0/QAHf6VJew/zhAHaeSCzkc4MrtShG7fgWo5Xj6Wcr8PBjenI+N685zLKOUPEXVwjXjdEl5/ffTebu54NZFpvJNYXMXz9wo3RBPZjid5/vUPKCKI5mljBnWnz8FVPNT5kkwlJGcVYFXgAd9gbozx/jwzXU4h8fQ4dH1zP4yi0pAObiDyc+upmN4DB0e/IzxMcc4q2Txwbx4XjgJFb8kYffwt/xsWq+huogVC4ztb5q/nUTTweqq9P3MfdF4v9NjsSxOKEahluTPPsfuwc3MeX8dzhb92LLzVTXkFJdRW1eHp6uzfObeDBL4Szn2445bYEd2Pkp1Bj8edeD2fjczyL8jCcdOkl+TxU/HdNzfvx/UnuLTD/awNfA2CmOfoHReTw6vT+IfydDBrTd/+z/3UxH7BOee8uDQ98lsyfVn7tLxvN8buvxBj2HdJMaZVluZXEaPx8M580o/BuRk8fpPeXD+N/66+Dj2U6ZQEftnMh7oyHvRifxg3s2tLaXD4EmUWPRjy3KKyyguv2D81Ri3rjb/XfotQb4G5DIdGdS3G5XfFbD3VwOb6jxYNUTH0EoPnHcUc+RoRxLt3HlriI7a9HT+maaQmBZPp1jz8h0YVwS4V7N//bc8eew8v5dUUUw3lGtM4zsN9ef+oC6g+POAZwbvF5+jOjGbD0srqV7xJe+tMDW078bJfPAFcOhO2Hg3TTyJeecqyC3W9q/GqEELr5Um6kCvgR4ExxawMaGO07cGEOwAuptdGVedyaYfHEn282SAC3AasOvIGy8/yqIRFl3UneHfrybyutsQkj4Ipmfmz4x6o6BJVRRdMB0h1Hnwnw+mcf9FxwtqSd51Y3+lNblQrZBbXEpNbR09u2n3V2PUIHOiK7AP8mRyt3Ms3VnO6L496QrYufszud95/nOkgoEDvegL2Pfz4l73Kt7ZuIMjpQqGijx+iklgZ90FisoMuLk442ZfxsHdeeyr79wVr5ugtvQC+TXXrsNhcHdmGIpY8PlBTl2AmrNZrF+9l8afmmIbcotLKSy7gJODDm837f5qjBok8Ffi0I87b9VBh85MHtDbeF8Hdwb5O1F4oQNht/Q13qcLZN78oTxZlcmIJ1Zj/9QPLC7rRC+dHw/d60Xl9h10fmIrXzi5UT8BsOvG7aFe9DxyGM8nvmfnNcqw6z6CZS/2I+TYIXr/KYauL+xmo86FG7uQ1Lrkl1ZYHJXX9q/GqEGVy2N1Mz/FEDtbjXqExtmFx6B8/hgAVTW1/HaygILS83Tv2plbvT3oqPEfkrhRMsKLdiunuJSC0vM46uzxce8qYVeBBF60SwVl5y86V76Hi3zVtBok8KLdqVZqyS0y/gCk/GqMuiTwot3JLS7jTGmF8Vdj3OVXY9QkgRftTo7FCTae8gOQqpLAi3bnQrWCg30H/D1kKq821eZKduExanUlNM7NuRO9XZ3lqHwLUG2LyufwQg124TGM7OvV1mXYLJnSC6EhEnghNEQCL4SGSOCF0BAJvBAaIoEXQkMk8EJoiAReCA2xosCnsvaF+axNbes6Gi913XxeWGdFBQubZ0WBF0LcKBVPVk5l7QuL2JRjujn6GTbOu4PtS2fytdebvP9wkPH+7e8RdnAYG+fdQeq6+SwwLeA7Yzq9N50mOPZF7rxW3wtnssnUN6nreWHhV2QDoOfZ2Be5k1TWvvAJTBnOvo+/IhsfZix+F6+vZ7Ii0bKdcQSOZjIj9q401d3wGKbHzfVhuc5oGOH9FZsSje1Zau4b8JnO2+8/BPXLLiJsk55nY+/l9AufQMS7PBJk/pvMt3/k3fCDeM04xaZNMGOx+b6VGLv1Md2n3rMlNEpRFMON/iPsn4aLpRjWzJ1nWJNiMBhS1hnmzl1nSDE9Ev/uA4Yl8QaDIX6pYUbYUkO8eYm18y66fTmLPg2X92uIX2qY8e52U7sHDDPMj8UvNcwIM63TtP65a1Ms1tnQZ8raeaY+jP83t7uo7pR1hrkW/V2qvt1lfVxS/0W3txuWhD1g0Xa7YYnltrj0b7VhhP3zhl+P8u/q/9S9HGn7e4R9nFR/c3QucOdwRvAJe1MfIijoRw7kTmfqPEhddwrfGU/Vj6ZBD09m9KaDjV9X7mmyc5JYEP5Vw30+XqTSG/BhRsRDBAF4e+HrM52pphV5e/lc1I3vjKfqR86ghyczOvwg27kDTueQnbiIsE0Wbb1SwRuw6A+4ZKZhakdTh2M9U8yzoNQ8TpJEYvhMVlg8ntuMXoWwpF7gt79H2NdevB37OUHA9qUzOQBAECEhEL0vlZB9WzkZ8pTx8dM59B52gy9f8zT7IqnsvbFeG7r/y+fMv3T/InX/JbfX88LC00yJ/dxiN6QxEVsAAAPsSURBVEEFpl0DCbhQk2oH7VLzToG3p+kF+iMHEhseC3p4Mr33/pcte3vVj2J3DtOT+PV6zMewU9dtJZEm8PbCN3HrDR+1z967/+IaRg/jTowzAcv6rir3NNk+Xngbe2Dv3pyrNAzCyzuHfftMPabuZ99Vm3rSO+crtmxv0p8ixHWpNsIbp8MrCQtfCegZPdry0TsI9l7JCp5hvvmuO1/k2YMz66fkvjOmM5rT11oDISGwoP6g3UO8/5fThC2ciXnW7TvjTd5/uGl1+3qfJjp8pnE67jOdt9+/w/T3vMuzp2da7DKYDpxd2sGd9zLj60Wmdj6MHt2wyxA0YjgsNB+0e5E7p0zn64Wm3QQfPaN9Lu3M7A7mL87jhYUzCfvYdNcVZzNCNE37+SEK09HviFacxhqn3081fIIg2pzlD1EI9bWb7xDa/vVXEPImQVh+HGV28cdlQojmabvAX3Jkm9HPsPHhICCI+bEydRWiJbSfKb0QyJS+pcmptUJoiGpT+gMnitTqSmhMcB/3ti5BM1QLvDxpQrR/MqUXQkMk8EJoiAReCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggNkcALoSESeCE0RAIvhIZI4IXQEAm8EBoigRdCQyTwQmiIBF4IDZHAC6EhEnghNEQCL4SGSOCF0BAJvBAaIoEXQkMk8EJoiAReCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggNkcALoSESeCE0RAIvhIZI4IXQEAm8EBqiSuCdnRwor6xRoyshRAtSJfAeLk7klVxQoyuhYUptXVuXYPNUCXy/nl1JO3VOja6EhqWdPkeAp0tbl2HTVAn8Hwb05LtfT6rRldCwbw/mMu4Wz7Yuw6apEviZoX34PCGTgnMyrRfNU6PUErM9lYfH9G3rUmyaKoEP8HThkbH9iPjHLjW6Exr0t42H6N/blXEDZYRvSap9LPf6zGEczSnhi12ZanUpNOJAZgGfxKWwYvaoti7F5qkWeCcHe2IiQnlu1W5+/O2UWt0KG3cgs4BHlv3Cssf1eLg4tXU5Ns9OURSDmh3+nJzH7OgEJgd787+PhuDs5KBm98JGKLV1vBl7kE/iUlj2uJ6wUf5tXZImqB54gPLKGl5Zf4CtB3J54o5A7tP7E+DlgpODvdqrElZEqa0jPa+Ubw7kELM9lf69XVkxe5SM7K2oRQJvtj+ziHU7M/nu0EnS80rlxApBgKcL427x5OExfeUAXRto0cALIdoXuXhGCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggN+f/sQGmHiJF2nwAAAABJRU5ErkJggg==)\n", - "\n", - "The procedure to create the cube is the following.\n", - "\n", - "1. On the stations file:\n", - " 1. replace empty states and countries in stations with a placeholder value (e.g., \"XX\");\n", - " 1. keep only the following fields: stationId, state, country, elevation\n", - "2. On the weather-sample file:\n", - " 1. filter out weather wrong measurements (i.e., where airTemperatureQuality=9);\n", - " 1. keep only the following fields: stationId, airTemperature, date, month, year\n", - " 1. create a new fulldate field by concatenating year, month, and date\n", - " 1. create a new fullmonth field by concatenating year and month\n", - "1. Join stations with weather measurements on the stationId field\n", - "1. Keep only the following fields: state, country, elevation, fulldate, fullmonth, year, airTemperature\n", - "1. Aggregate the measurements by state and date to take the average temperature\n", - " - Group by: state, country, elevation, fulldate, fullmonth, year\n", - " - Calculation: avg(airTemperature)\n", - "1. Save the result on a file" - ], - "metadata": { - "id": "Dv5BUnaAWWEi" - } - }, - { - "cell_type": "code", - "source": [ - "!wget https://big.csr.unibo.it/downloads/bbs-dm/datasets/weather-stations.csv\n", - "!wget https://big.csr.unibo.it/downloads/bbs-dm/datasets/weather-sample-10k.txt" - ], - "metadata": { - "id": "REn3hjicbRFL" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "## Spark" - ], - "metadata": { - "id": "xO6bQ3NQwDAj" - } - }, - { - "cell_type": "code", - "source": [ - "dfW = sc.textFile(\"weather-sample-10k.txt\")\\\n", - " .map(lambda l: (l[4:15],l[15:19],l[19:21],l[21:23],int(l[87:92])/10,l[92:93]))\\\n", - " .toDF([\"stationId\",\"year\",\"month\",\"day\",\"airTemperature\",\"airTemperatureQuality\"])\n", - "dfW.show()" - ], - "metadata": { - "id": "Xcd4fAhgeDhH" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "code", - "source": [ - "from pyspark.sql.functions import concat\n", - "dfS = spark.read.option(\"delimiter\", \",\").option(\"header\", \"false\").csv(\"weather-stations.csv\")\n", - "dfS = dfS.select(concat(dfS[0],dfS[1]),dfS[2],dfS[3],dfS[4],dfS[5],dfS[6],dfS[7],dfS[8],dfS[9],dfS[10])\\\n", - " .toDF(\"stationId\",\"city\",\"country\",\"state\",\"call\",\"latitude\",\"longitude\",\"elevation\",\"date_begin\",\"date_end\")\n", - "dfS.show()" - ], - "metadata": { - "id": "OaknyiXHeCbF" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "1. On the stations file:\n", - " 1. replace empty states and countries in stations with a placeholder value (e.g., \"XX\");\n", - " 1. keep only the following fields: stationId, state, country, elevation" - ], - "metadata": { - "id": "8QqJZrglm-ZD" - } - }, - { - "cell_type": "code", - "source": [ - "dfS1 = dfS.fillna({'state': 'XX', 'country':'XX'})\n", - "dfS2 = dfS1.select('stationId','state','country','elevation')\n", - "dfS2.show()" - ], - "metadata": { - "id": "8v1YuN-4nQCQ" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "2. On the weather-sample file:\n", - " 1. filter out weather wrong measurements (i.e., where airTemperatureQuality=9);\n", - " 1. keep only the following fields: stationId, airTemperature, date, month, year\n", - " 1. create a new fulldate field by concatenating year, month, and date\n", - " 1. create a new fullmonth field by concatenating year and month" - ], - "metadata": { - "id": "5LxL1WQvnF0J" - } - }, - { - "cell_type": "code", - "source": [ - "from pyspark.sql.functions import concat, lit\n", - "dfW1 = dfW.where(\"airTemperature < 9\")\n", - "dfW2 = dfW1.select('stationId','airTemperature','day','month','year')\n", - "dfW3 = dfW2.withColumn(\"fulldate\", concat(dfW1.year,lit(\"-\"),dfW1.month,lit(\"-\"),dfW1.day))\n", - "dfW4 = dfW3.withColumn(\"fullmonth\", concat(dfW1.year,lit(\"-\"),dfW1.month))\n", - "dfW4.show()" - ], - "metadata": { - "id": "Cf9bm6CspR8x" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "3. Join stations with weather measurements on the stationId field\n" - ], - "metadata": { - "id": "jy2yp35unHhO" - } - }, - { - "cell_type": "code", - "source": [ - "dfJ = dfS2.join(dfW4, \"stationId\")\n", - "dfJ.show()" - ], - "metadata": { - "id": "oHTKQhQJrPMU" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "4. Keep only the following fields: state, country, elevation, fulldate, fullmonth, year, airTemperature" - ], - "metadata": { - "id": "mv_dfniznJ2A" - } - }, - { - "cell_type": "code", - "source": [ - "dfJ2 = dfJ.select(\"state\", \"country\", \"elevation\", \"fulldate\", \"fullmonth\", \"year\", \"airTemperature\")\n", - "dfJ2.show()" - ], - "metadata": { - "id": "dYyMxZaSrprK" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "5. Aggregate the measurements by state, country and date to take the average temperature\n", - " - Group by: state, country, elevation, fulldate, fullmonth, year\n", - " - Calculation: avg(airTemperature)" - ], - "metadata": { - "id": "YXKChhXwnLlv" - } - }, - { - "cell_type": "code", - "source": [ - "dfG = dfJ2.groupBy(\"state\", \"country\", \"elevation\", \"fulldate\", \"fullmonth\", \"year\").agg({'airTemperature': 'avg'})\n", - "dfG.show()" - ], - "metadata": { - "id": "shVr9k9hryqH" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "6. Save the result on a file" - ], - "metadata": { - "id": "zwXx8h-tv4-v" - } - }, - { - "cell_type": "code", - "source": [ - "dfG.write.mode('overwrite').option('header','true').csv(\"weather-cube\")" - ], - "metadata": { - "id": "pKTH93LQu61U" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "## PowerBI\n", - "\n", - "Download the file from the left panel of this notebook (in case of issues, download it from [here](https://raw.githubusercontent.com/w4bo/2023-bbs-dm/master/materials/datasets/weather-cube.csv)) and load it in [Power BI](https://app.powerbi.com/).\n", - "- Visualize the daily trend of average temperatures for each country\n", - "- Show the average temperature on the map\n", - "- Compute bins for the elevation field and show the average temperature for each bin\n", - "\n", - "\n", - "The final Power BI file will be available [here](https://big.csr.unibo.it/downloads/bbs-dm/results/weather-cube.twb)." - ], - "metadata": { - "id": "jkKWX-mWv8S9" - } - }, - { - "cell_type": "markdown", - "source": [ - "# Additional exercises\n", - "\n", - "The solution will be available [here](https://big.csr.unibo.it/downloads/bbs-dm/results/2024_bbs_dm_spark_basics_solution.ipynb).\n", - "\n", - "## Getting familiar with data frame transformations\n", - "\n", - "Carry out the following operations (in any order).\n", - "\n", - "- ```.select()``` operator; starting from ```dfS```:\n", - " 1. keep only country, elevation, date_begin, and date_end\n", - " 1. keep only the first four characters of date_begin using [sf.substring](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.substring.html)\n", - " 1. use [concat](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.concat.html) to concatenate date_begin with date_end and putting an underscore (\\_) in the middle; since the underscore is not a column, declare it as ```sf.lit(\"_\")```\n", - " 1. use [coalesce](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.coalesce.html) to take the value of country if not null, otherwise the value of city\n", - " 1. put countries in lowercase using [sf.lower](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.lower.html)\n", - " 1. in the previous four points, use ```.alias()``` to give a meaningful name to the obtained columns\n", - "- ```.withColumn()``` operator; starting from ```dfS```:\n", - " 1. do the same as points 2 to 5 of the ```select``` operator, but using ```withColumn```\n", - "- ```.filter()``` operator; starting from ```dfS```, keep only the rows where:\n", - " 1. elevation is greater than 5000\n", - " 1. country is not null, using [sf.isnotnull](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.isnotnull.html)\n", - " 1. conditions 1 and 2 are both true; conditions must be put between parenthesis and separated by \"&\" (e.g., check [here](https://www.geeksforgeeks.org/pyspark-filter-dataframe-based-on-multiple-conditions/))\n", - " 1. either one of conditions 1 and 2 is true; conditions must be put between parenthesis and separated by \"|\" (e.g., check [here](https://www.geeksforgeeks.org/pyspark-filter-dataframe-based-on-multiple-conditions/))\n", - " 1. date_begin is the first day of the month (requires to use substring)\n", - "- ```.groupBy()``` operator; starting from ```dfW```:\n", - " 1. group by airTemperatureQuality to count how many rows there are for each value\n", - " 1. as above, but also calculate the average temperature\n", - " 1. as above, but also given meaningful names to the results using ```withColumnRenamed```\n", - " 1. group by month to calculate the minimum and maximum temperatures and order by month using ```orderby```; to aggregate differently on the same column, use the [sf.max](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.max.html) and [sf.min](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.min.html) functions inside the ```agg``` function instead of the object enclosed by brackets ```{}```\n", - " 1. group by month and day to calculate the minimum and maximum temperatures\n", - " 1. group by stationId and year month to count the number of rows" - ], - "metadata": { - "id": "679cmN5vLOMH" - } - }, - { - "cell_type": "code", - "source": [ - "import pyspark.sql.functions as sf\n", - "\n", - "# TODO HERE" - ], - "metadata": { - "id": "82YodmWqG-JZ" - }, - "execution_count": null, - "outputs": [] - }, - { - "cell_type": "markdown", - "source": [ - "## Complete exercises\n", - "\n", - "Carry out the following exercises (in any order).\n", - "\n", - "- Check if there exist stations with a negative elevation; then, calculate how many of these stations exist in each country; rename the result to \"cnt\" and order the result by decreasing cnt\n", - "- Take only stations with positive elevation, compute the maximum elevation by country and rename the result to \"elevation\"; then, join the result with the original dfS to get, for each country, the name of the city with the highest elevation (join key: ```[\"country\",\"elevation\"]```); order the result by decreasing elevation\n", - "- Take only weather values with airQuality==1, compute the minimum temperature for each stationId and rename it to \"minTemperature\"; then, join the result with dfS and keep only the columns \"minTemperature\" and \"elevation\"; finally, use the correlation between the two columns. To do the last part, you need to:\n", - " - cast the elevation to an integer datatype: you need to add ```from pyspark.sql.types import IntegerType``` and then ```df.myfield.cast(IntegerType())```;\n", - " - compute the correlation with ```df.stat.corr(\"myfield1\",\"myfield2\")```." - ], - "metadata": { - "id": "JFGe5AvcHCKM" - } - }, - { - "cell_type": "code", - "source": [ - "# TODO HERE" - ], - "metadata": { - "id": "KoSvRLcWYXQq" - }, - "execution_count": null, - "outputs": [] - } - ] -} \ No newline at end of file +{"nbformat":4,"nbformat_minor":0,"metadata":{"colab":{"provenance":[{"file_id":"1pOVAYbnaLc_KEkrop-dwD3qlAZYKwjCr","timestamp":1711709014150}],"authorship_tag":"ABX9TyOBxrWRv/pd7+23sx/7cFes"},"kernelspec":{"name":"python3","display_name":"Python 3"},"language_info":{"name":"python"}},"cells":[{"cell_type":"markdown","source":["# Install Spark & initialize application\n","\n","Run the following code to install Spark in your Colab environment."],"metadata":{"id":"EsElqAaj4Sse"}},{"cell_type":"code","execution_count":null,"metadata":{"id":"fbvEUbWIHm2s"},"outputs":[],"source":["!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n","!wget -q https://archive.apache.org/dist/spark/spark-3.5.1/spark-3.5.1-bin-hadoop3.tgz\n","# If the above command is too slow, uncomment the following and try it\n","# !wget -q https://big.csr.unibo.it/downloads/bbs-dm/spark-3.5.1-bin-hadoop3.tgz\n","!tar xf spark-3.5.1-bin-hadoop3.tgz\n","!pip install -q findspark"]},{"cell_type":"code","source":["import os\n","os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n","os.environ[\"SPARK_HOME\"] = \"/content/spark-3.5.1-bin-hadoop3\"\n","import findspark\n","findspark.init()\n","findspark.find() # Should return '/content/spark-3.5.1-bin-hadoop3'"],"metadata":{"id":"4oTFM5YtJvv7"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["from pyspark.sql import SparkSession\n","\n","spark = SparkSession.builder\\\n"," .master(\"local\")\\\n"," .appName(\"Colab\")\\\n"," .config('spark.ui.port', '4050')\\\n"," .getOrCreate()\n","sc = spark.sparkContext\n","\n","sc"],"metadata":{"id":"KJlzVAmbJ9vL"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["# Spark: working with RDDs\n","\n","Check the documentation: [here](https://spark.apache.org/docs/latest/api/python/reference/pyspark.html#rdd-apis)."],"metadata":{"id":"oBd7XwkFBDEF"}},{"cell_type":"markdown","source":["## Basics"],"metadata":{"id":"jcoFwGvm4gj6"}},{"cell_type":"code","execution_count":null,"metadata":{"autoscroll":"auto","id":"vlFswJyWytKG"},"outputs":[],"source":["# let's create a simple example\n","riddle1 = \"over the bench the sheep lives under the bench the sheep dies\"\n","riddle2 = [\"over the bench the sheep lives\", \"under the bench the sheep dies\"]"]},{"cell_type":"code","execution_count":null,"metadata":{"autoscroll":"auto","id":"WHywj4BmytKH"},"outputs":[],"source":["# create an RDD from the `riddle` string\n","rdd1 = sc.parallelize(riddle1.split(\" \"))\n","# each tuple of the RDD corresponds to a single word\n","\n","print(rdd1)\n","# why is there no result returned?"]},{"cell_type":"code","execution_count":null,"metadata":{"autoscroll":"auto","id":"3oYjBLnOytKI"},"outputs":[],"source":["# compute the RDD\n","print(rdd1.collect())"]},{"cell_type":"code","source":["rdd2 = sc.parallelize(riddle2)\n","print(rdd2.collect())"],"metadata":{"id":"exzQLruZ9qgg"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["## Transformations"],"metadata":{"id":"iyH7halU9HiB"}},{"cell_type":"code","source":["# map: returns a new RDD by applying a function to each of the elements in the original RDD\n","rdd1.map(lambda s: s.upper()).collect()"],"metadata":{"id":"sndBHyEF86T5"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# flatMap: returns a new RDD by applying the function to every element of the parent RDD and then flattening the result\n","rdd2.flatMap(lambda s: s.split(\" \")).collect()"],"metadata":{"id":"9MlPRBd_-Cl1"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# filter: returns a new RDD containing only the elements in the parent RDD that satisfy the function inside filter\n","rdd1.filter(lambda s: s.startswith(\"u\")).collect()"],"metadata":{"id":"UlT_jxmH9Myx"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# distinct: returns a new RDD that contains only the distinct elements in the parent RDD\n","rdd1.distinct().collect()"],"metadata":{"id":"QxxJdRxW-Xcj"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# groupByKey: groups the values for each key in the (key, value) pairs of the RDD into a single sequence\n","rdd1.map(lambda s: (s,1)).groupByKey().mapValues(list).collect()\n","\n","# (first map converts to a key-value RDD)\n","# (mapValues is a map that operates only on the values - in this case, used to convert from ResultIterable to List for printing reasons)"],"metadata":{"id":"dBAh2Gs8-fdM"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# reduceByKey: when called on a key-value RDD, returns a new dataset in which the values for each of its key are aggregated\n","rdd1.map(lambda s: (s,1)).reduceByKey(lambda x, y: x + y).collect()"],"metadata":{"id":"cH1dxiGl_cS5"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# sortByKey: returns a new RDD with (key,value) pairs of parent RDD in sorted order according to the key\n","rdd1.map(lambda s: (s,1)).sortByKey().collect()"],"metadata":{"id":"-XJWbbv6_0Tw"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# join: starting from two RDD with (key, value1) and (key, value2) pairs, returns a new RDD with (key, (value1, value2)) pairs\n","\n","# rddProvinces: (initials, name, region)\n","rddProvinces = sc.parallelize([(\"BO\", \"Bologna\", \"Emilia-Romagna\"),(\"RA\", \"Ravenna\", \"Emilia-Romagna\"),(\"MI\", \"Milan\", \"Lombardia\")])\n","# rddPeople: (id, name, province)\n","rddPeople = sc.parallelize([(1, \"Enrico\", \"RA\"),(2, \"Alice\", \"RA\"),(3, \"Bob\", \"BO\"),(4, \"Charlie\", \"FC\")])\n","\n","# This does not work\n","rddPeople.join(rddProvinces).collect()"],"metadata":{"id":"SGIn90U0Md8j"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["rddProvinces2 = rddProvinces.map(lambda p: (p[0], (p[1],p[2])))\n","rddProvinces2.collect()"],"metadata":{"id":"gQ54N0nKjmJC"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["rddPeople2 = rddPeople.map(lambda p: (p[2], (p[0],p[1])))\n","rddPeople2.collect()"],"metadata":{"id":"PmG1337hk0a9"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["rddPeople2.join(rddProvinces2).collect()"],"metadata":{"id":"AJaZHHXYj-vp"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["## Actions"],"metadata":{"id":"gKJ_7MCsAIy_"}},{"cell_type":"code","source":["# collect: returns a list that contains all the elements of the RDD\n","rdd1.collect()"],"metadata":{"id":"XQXX-d20_vWo"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# count: returns the number of elements in the RDD\n","rdd1.count()"],"metadata":{"id":"FXuL9K2qATfL"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# reduce: aggregates the elements of the RDD using a function that takes two elements of the RDD as input and gives the result\n","sc.parallelize([1, 2, 3, 4, 5]).reduce( lambda x, y: x * y)"],"metadata":{"id":"YI10eDDnAWe7"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# take: returns the first n elements of RDD in the same order\n","rdd1.take(2)"],"metadata":{"id":"CgN5qtwBAvIv"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# saveAsTextFile: saves the content of the RDD to a file\n","rdd1.saveAsTextFile(\"rdd1\")"],"metadata":{"id":"yS8SuYRQvV6F"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["## Examples"],"metadata":{"id":"gmosJNIOA5S_"}},{"cell_type":"code","execution_count":null,"metadata":{"autoscroll":"auto","id":"mGRkO7_oytKK"},"outputs":[],"source":["# Flatten the words beginning with the letter C\n","# - map: transform each string in upper case (remember: map returns a new RDD with the same cardinality)\n","# - filter: keep only the strings beginning with \"C\" (remember: filter returns a new RDD with the same or smaller cardinality)\n","# - flatMap: explode each string into its characters (remember: flatMap returns a new RDD with the any cardinality)\n","rdd1\\\n"," .map(lambda s: s.upper())\\\n"," .filter(lambda s: s.startswith(\"U\"))\\\n"," .flatMap(lambda s: list(s))\\\n"," .collect()"]},{"cell_type":"code","execution_count":null,"metadata":{"autoscroll":"auto","id":"Wrr37cZ2ytKL"},"outputs":[],"source":["# A simple word count\n","# - map: map each word to a tuple (word, 1); each tuple represent the count associate with a word\n","# - reduceByKey: group all the tuples with the same word and sum the counts\n","# - sortBy: sort tuples by count\n","rdd1\\\n"," .map(lambda s: (s, 1))\\\n"," .reduceByKey(lambda a, b: a + b)\\\n"," .sortBy(lambda x: x[1], False)\\\n"," .collect()"]},{"cell_type":"code","source":["# Compute average length of words depending on their initial letter\n","# map: map each word to a key-value tuple (word, (wordLength, 1)), where the value is an object composed by two value: the length of the word and a 1\n","# reduceByKey: group all the tuples with the same word to 1) sum the lengths, and 2) sum the counts\n","# mapValues: divides the sums by the counts to compute the averages\n","# sortBy: sort tuples by averages\n","rdd1\\\n"," .map(lambda s: (s[0], (len(s),1)))\\\n"," .reduceByKey(lambda a, b: (a[0] + b[0], a[1] + b[1]))\\\n"," .mapValues(lambda x: x[0]/x[1])\\\n"," .sortBy(lambda x: x[1], False)\\\n"," .collect()"],"metadata":{"id":"zxBoMsnTCATh"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["# Spark: working with DataFrames\n","\n","Check the documentation: [here](https://spark.apache.org/docs/3.2.1/api/python/reference/api/pyspark.sql.DataFrame.html).\n","\n","What is different from Pandas' DataFrames?\n","\n","- Spark supports parallelization (Pandas doesn't), thus it's more suitable for big data processing\n","- Spark follows Lazy Execution, which means that a task is not executed until an action is performed (Pandas follows Eager Execution, which means task is executed immediately)\n","- Spark has immutability (Pandas has mutability)\n","- The data structure is similar, the APIs are different"],"metadata":{"id":"sCXrRXAAEtet"}},{"cell_type":"code","source":["!wget https://raw.githubusercontent.com/w4bo/2024-bbs-dm/master/materials/datasets/housing.csv\n","df = spark.read.option(\"delimiter\", \",\").option(\"header\", \"true\").csv(\"housing.csv\")\n","df.show()"],"metadata":{"id":"IGChYOQWRZku"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# Switching from Spark to Pandas\n","pandasDF = df.toPandas()\n","print(pandasDF)"],"metadata":{"id":"_aSykf1wV8dB"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# Switching from Pandas to Spark\n","df = spark.createDataFrame(pandasDF)\n","df.show()"],"metadata":{"id":"cFUtMUaUWTY5"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# select: returns a new DataFrame with only selected columns (similar to a map on RDDs)\n","df.select('population','median_house_value').show()"],"metadata":{"id":"eFAs2zBHPLTA"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# select, similarly to a map, allows column values to be redefined\n","df.select(df.population,df.median_house_value/1000).show()\n","# put the operation within parenthesis and add .alias('median_house_value_in_K$')"],"metadata":{"id":"FmClDtFaQqO7"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# withColumn: used to manipulate (rename, change the value, convert the datatype)\n","# an existing column in a dataframe (or to create a new column) while keeping the rest intact\n","df.withColumn('median_house_value_in_K$',df.median_house_value/1000).show()"],"metadata":{"id":"4whu8BylWfAK"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# filter: returns a new DataFrame containing only the elements in the parent DataFrame that satisfy the function inside filter (as in RDDs)\n","# orderBY: orders the DataFrame by the selected column(s)\n","df.filter(df.population > 1000).orderBy(df.population.asc()).show()"],"metadata":{"id":"9zrnJn3_Q0PD"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# groupBy: returns a new DataFrame which is the result of an aggregation\n","df.groupBy(df.ocean_proximity).agg({'median_house_value': 'avg', '*': 'count'}).show()"],"metadata":{"id":"1qv5_B90Raho"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# withColumnRenamed: rename a column\n","df.groupBy(df.ocean_proximity).agg({'*': 'count'}).withColumnRenamed(\"count(1)\", \"tot\").show()"],"metadata":{"id":"E3b1rxzMN7Lm"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["# SQL queries can be run on a DataFrame\n","df.createOrReplaceTempView(\"housing\")\n","spark.sql(\"select ocean_proximity, avg(median_house_value) as avg_price from housing group by ocean_proximity order by avg_price desc\").show()"],"metadata":{"id":"TaP-GatdTD7k"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["# Exercise: creating a cube\n","\n","You are working with two files:\n","\n","- weather-stations.csv: it contains a list of weather stations that capture weather information every day of every year throughout the world\n"," - Each station is identified by a StationID\n","- weather-sample-10k.csv: it contains the data measured by a certain station on a certain date (a sample of 10k lines collected from the National Climatic Data Center of the USA)\n"," - Each weather measurenent is identified by a StationID and a Timestamp\n","\n","Your goal is to create a single file representing the following cube and to run some queries through PowerBI.\n","\n","![image.png](data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAPwAAAEICAYAAAB735ncAAAgAElEQVR4nO3deVyU5f7/8RcyICoiASoomwqklhuKo6HHSu1Y5pJQtp1jZd+Kk5Sl/fppx6I62befWZkZns7Bkx23TmhpR1sIWxQR9zRRVglQkU1kUZYb5vfHzMC4A94sM/fn+Xj4eDgz133dH+6Z91zXfc99z9gpimJACKEJHdq6ACFE65HAC6EhEnghNEQCL4SGSODbrXiedwzl4zTDJffNIU4Os4pmksC3W+OJXN6BtXHp1Oc7fgv7P3ieCXZtWZewZhL4dizg6ShGzF3GDwaAdP4etY+HJwYgeRfNJYFv18YzNeIToqLTMMQvJ3JIFBGBdhin9jp0Oh06ncW0P/55HHVXuX9ONNFjHdCFfkya7BJolgS+nRsfuZwOa5fx3Gv7+OD5CdgRz/OOX3JvlYKiKCjH/8z6x6KNIR6/jGrFdP93I5i77IeG3YGVazCsqkZJ+AuBMkXQLF1bFyCuI+BpokY48sfardQE2kF6OgcMK1nhsNKi0TNkAYGk8/exA3g20RRz/c2kGyYSCPCMeXYgtExGeCvgN1CPfqB/wx0jP+RYjWkkVxQU5SMm2hnDfuTVatPIv4LRbVeyaKck8NYmIIDgvc+x7IdLd8R/JzlpJOb3hfT4Nexu9eJEeydTeqsznmXJKxg7wAGdOfPPfEPNRxOJXB7FgAEOPAfon4lAf52eNu/NZldKPr/lnCXlVCn+3Z3RB3ZnZIAHk4O90dnLeGBr7ORqOe3Jyi/nyZUJKHV1TAvxY6i/GwGe3cgtqmBfRgHxR06TnlfK6jljGd7Xva3LFSqSwGvM2h0ZzPtsLy9PH8z8qYOu2m7NL+m89NkeFtw3mGcnDWjFCkVLksBryPGTJYxZtI3ExVPo39v1uu3zSi4wbP6XbH55goz0NkJ20jRCqa3jyZW7eO3+YY0KO4Cnayeinwpl1kc7qKypbeEKRWuQwGvE1gO5ODnYM/feW5u03PSRfowK6kFMfGoLVSZakwReI/akFzJ+UK9mLTtuoCd70gtVrki0BQm8RiSknGFoH7dmLTu0jxtJaQUqVyTaggReI7Lyyxnq37wDb0P93UnPK1W5ItEWJPAa4d/DmeMnS5q1bHpeKd7uXVSuSLQFCbxG6AO6cyiruFnLHjpRhD6wu8oVibYggdeI4L7u/Hw0r1nL/pycx8gAD5UrEm1BAq8RYXo/sgrK2JCQ2aTldh47Q2ziCWaNC2ihykRrksBrhM6+A6vnjCUyJpHcoopGLVNeWcPjK35h2eN6PFycWrhC0Rok8BoyxM+NBdMHMXbRVmITT1yz7U9HTzPoxS+ZFuJL2Cj/a7YV1kPOpdeghOP5zFqxgzH9ezItxJehfdwJ8HQxXS1XSPyRU/x3fw7LHh/J5GCfti5XqEgCr1HllTWs3ZHBL8lnOJRVTMqpc/Ry60xIPw9GBnjw9MT+uHZxbOsyhcok8IIL1Qo/H88GYNLgvm1cjWhJsg8vhIZI4IXQEAm8EBoigRdCQyTwQmiIBF4IDZHAC6EhEnghNEQCL4SGSOCF0BAJvBAaIoEXQkMk8EJoiAReCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggNkcALoSESeCE05OqBrzrMkuE6dDrjP69Xd7diWTYqez33uer401elV3hwD1HeV3us+fZEeavep7BeVwx81eElDO/yR8pWKCiK8V/aqIOs/b1liijfPEs7byi36Mn873YujWD2+reJ89PT8Ub73xOF95++uqx/IeCKgS/n26XvMXnnad4Y1XCv8z0RPOLXeoXZrgEEZy8h+tdKi/uySfiPO/c/2qPNqhLacHngy+PZ9P3/cM+oK7Q2N9k8q36qr3OZWj/yXzpSW942/j+OzbNMy3m9ym7T/a5haylYPMbUVxWHl4QyNXoZs3Q6XEJDGW5qC5h2Nbyw3glBIJMe6sGWb1Ooj/yeVbzkfC+3nNlLlUXLPVHeDds5+B3M7xHZ6+8j9J2t/Os+14sf2xOF921/I299OG66YN6pf1NJIsrb1I+M/pp2WeCrMlNJD7kZr6ssUL55Fq7P+LHTNNWv2DmW90ZZBPIaChb/mf0RCopSwsa7PmDx2t9xnraako2P0H3hTpTSLaZZRCXb3igiQlEoTfiOV+/6B9tMK6hK+Y6ver/L7Gu8IbV33g8tYOLyd/nWlLw923YROX8S3Sza7Iny5raMjyg2befMl3Yz7n8awnr0lUXkR+WhKJn82+9t3v22FEZGkbvrr3g+FEuxcoCXhzgBsD78M4YmKyjKLv764xuXzC6EllwW+I59gwjYm8LpKzavIjP1FAu/fANz3joOjrgokNfSfeGXpt0EZ8bPuI+9KVdeCzhxz7uzTetwZvyMu/jHtt1AFSnffYX7zDFY997FSO558kc2bi+F7PW8HTeeSTc7WTyeTcavk4hdMR0X0z2+Dy1g7o8b2W5K/C1v/YvnhzgBvoQ+cAeZKSe4Wowfil3BdBfzevP57UR1S/1hop27fErv0IdbezUuwK3FedI8Xkzcxu6qFL77fjwLbeBgwsgnllC+JJqt8f/B/aUIhjhdfxkhbtTlge84mIhX72LxXVMtjspXcXjZS6z9vSN9g3qx+L6GKXzV4WjeMO3zO/S5lV7/2GZ6rJz4TWvVqbLjzfzxrkQ2rvqO70ffgxXP5hv4hvJAj7eZ9j/O3Huny6UP0m/It4Q/2zCFz17/Nh/cEcZlTYVoAt2V7nSetpqKnUu4rZ+OWQB05p7VR9jiB/h9woHU2wjWLTY27nwPq49sMYZwcASv3uXKGN1ioDsREY80qgjn8TO4K2wMug/uYfWRLxh0WYuO3PzHu9g+ZgcvHnmp/t7Ne7PZlZLPbzlnSTlVin93Z/SB3RkZ4MHkYG909u35vCJfHlowl3+Nuo9JVwjxyKhd/Pu+wbjpyo13DH6L/buMU/ySa3U78h6e/PE23HSDeWv/Lu64RlPz9jucfZajOWfxdHVix9ESK9l+ojnsFEUxtHURjVF1eAl3/rUX67Y8giG/nCdXJqDU1TEtxI+h/m4EeHYjt6iCfRkFxB85TXpeKavnjGV4X/e2Lr3dyZLtp1lWEvhyNs/qRcxdR5jpqzDvs728PH0w86dePhcwW/NLOi99tocF9w3m2UkDWrHW9m3tjgzZfhrW7gNfdXgJtwUv4NTCnfz4dH/GLNpG4uIp9O/tet1l80ouMGz+l2x+eYKMVMDxkyWy/TSu3QfeTKmt4/aob5l5Wx/m3ntro5f7as/vLFi7j33vTMHJwb4FK2zfZPsJsKKr5bYeyMXJwb5JL1aA6SP9GBXUg5j41BaqzDrI9hNgRYHfk17I+EG9mrXsuIGe7EkvVLki6yLbT4AVBT4h5QxD+7g1a9mhfdxISitQuSLrIttPgBUFPiu/nKH+zTtwNNTfnfQ8bV8yIttPgBUF3r+HM8dPXvOUk6tKzyvF272LyhVZF9l+Aqwo8PqA7hzKKm7WsodOFKEP7K5yRdZFtp8AKwp8cF93fj6a16xlf07OY2SAh8oVWRfZfgKsKPBhej+yCsrYkJDZpOV2HjtDbOIJZo0LaKHKrINsPwFWFHidfQdWzxlLZEwiuUUVjVqmvLKGx1f8wrLH9Xi4aPv6U9l+Aqwo8ABD/NxYMH0QYxdtJTbxxDXb/nT0NINe/JJpIb6EjfJvpQrbN9l+wmpOrbWUcDyfWSt2MKZ/T6aF+DK0jzsBni6mq70KiT9yiv/uz2HZ4yOZHOzT1uW2O7L9tMsqAw/G6ebaHRn8knyGQ1nFpJw6Ry+3zoT082BkgAdPT+yPaxfHti6z3ZLtp01WG3hLF6oVfj6eDcCkwX3buBrrI9tPO6xqH14IcWMk8EJoiAReCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggNkcALoSESeCE0RAIvhIZI4IXQEAm8EBoigRdCQyTwQmiIBF4IDVE18GdjH0Sn0xn/9XuL/Wp2LlrW2Z18PCmo4fnT6eg37yea9+NUor1SLfDn97/F7S8PIklRUBSF0v+48dt+gLPEPtiPt66b/vPsfyuUmbG5apUkmqrnY/yQb3z+FCWPlVWz6DE+ht+vs1hh3BxC5c3BKqgW+KoTRyh/YhLDTbc7D49g1vBrLiLaNQ8mfrSfbwas4sOfJMq2QrXA3zQ+nJDX7rpkhD5L7IPdeTD2d17T63CfGUvu+f28NcRi2vjWfoyj+2j0ryWx8UH/ht2Bs7E8aJ5ius9EBv/W5sGwacPY/fVhiikkbo5fw5TfNPIXxs1h+N0rSVo2gR668cT8DlzSdnzM9eYIorWotw9/UzgbSr9n8Jv+6HTmKfxNhG8oYEO4H68nKRR9Ho535+G88qtp2liaxBPb/h+xuZ0Z/koiSa/rCduQhZLxCsPPxvLgiBReMu8ifD+Y9+fFIplvKx5M/Oj3+um+eeT3mPgR+795Bv3zP5CvxDPbr5C4OdP4Nvygqe1xwlc9gWS+fVD3KL05zAXvcER/9f32+oN7LnpeS9pHypnL25zPTOHY76+hN40SLvrXSNqXwhWaihbmM7APbgDp0YTqdOh0nty9MolDGWVXaF1C+sEklk3oYRrh+xOZlEXyCdktaA9a5mO5m8KJ3hDCqm8vT/zZ2AcJ+CKcLEVBUQrYEH71bpzDNpjamf5lvIIcFmhN6XwR9Q29+3U1hv2P6bxlOqh3fLn+GsuN4+8ZFs+bksHS291arWpxdaoF/mxspMWIfp7MlGNXaGW833VwH7wBzsYTG3vl/jr3vRmfjS+zWj7bayPpRIf2591RMbxyuxuFJ46S6z+QPsahnrg1SVdZzpWAYeks/lCO2rdHOrU6uumex2C0Dt1R4+1uYRv49XPjeDw+PIQH9TreC9vArysXEdJdj+41wC+CiPoRvjMDJt3Dcb0/Or/XScp4hVVJKYzW63jN1MLv9SQyXmkY4zfvzWZXSj6Hs89yNOcsnq5O7DhawsgADyYHe6Ozl/OKruWi7ZeWRk3Wp0zo8Wn94+P+nkHGbD/jjYlz+b9R/emnexoYR0REwwjvMWwaw56cQI9l4/h7RjyzP4pnfmh/ephfXX7P88Pepcgg3/as8scks/LLeXJlAkpdHdNC/Bjq70aAZzfTzx0XEH/kNOl5payeM5bhfd3butx2R7afdlld4NfuyGDeZ3t5efpg5k8ddNV2a35J56XP9rDgvsE8O2lAK1bYvsn20zarCvzxkyWMWbSNxMVT6N/b9brt80ouMGz+l2x+eYKMVMj2E1Z08YxSW8eTK3fx2v3DGvViBfB07UT0U6HM+mgHlTW1LVxh+ybbT4AVBX7rgVycHOyZe++tTVpu+kg/RgX1ICY+tYUqsw6y/QRYUeD3pBcyflCvZi07bqAne9ILVa7Iusj2E2BFgU9IOcPQPs37XGdoHzeS0gpUrsi6yPYTYEWBz8ovZ6h/8w4cDfV3Jz2vVOWKrItsPwFWFHj/Hs4cP1nSrGXT80rxdu+ickXWRbafACsKvD6gO4eymney5qETRegDu6tckXWR7SfAigIf3Nedn4/mNWvZn5PzGBngoXJF1kW2nwArCnyY3o+sgjI2JGQ2abmdx84Qm3iCWeMCWqgy6yDbT4AVBV5n34HVc8YSGZNIblFFo5Ypr6zh8RW/sOxxPR4uTi1cYfsm26850ogO1V30xZ6RcQBxROoiiWuFdUeqvBKrCTzAED83FkwfxNhFW4lNPHHNtj8dPc2gF79kWogvYaP8W6nC9k22X3PoWX6s4dr+5RNbbk1p0aGERqeZbgUSkaD++lS7PLa1PD/5Fkb0686sFTvYvDebaSG+DO3jToCni+lqr0Lij5ziv/tzWPb4SCYH+7R1ye2KbD9ts6oR3iy0fw8OLZnK6CAPYhNPMOXtOOzCY9Av2MKH247g5erEvv+dIi/Wq7Dcfut3ZjDxjW+wC48h9K//5dMfU2X7NVWa+au/dOhMU/206FB0lvPxtGhCQ6NJu3Q3wdQmLTqUAZFJJEUOMPWRRnRoKPUD/kXrsJzqm9rFNTzeMEu4nNWN8GbOTg48PbE/T0/sD0BucRm/5RbQpaMDY2+WF+r1mLffjFG+7D9hPHo/oo8nHl07t3Fl7U0SkQN0RALG6X0CEYEWD6dFE/oYfKooBALERaKLjENZHkWEbgtxyycyEUj7YQ08+imBBBKYoBBhXJjo0MeITptIREQCxwjlMT4lISIQSCPdch0D1vDoMYUE40qI1IUSXV9LEpFRj3JMUQhMiyZ0wIfERSznSnsDVjnCX4mDzvin1NTWtXEl1kWx2F4O9vZtWEl7ZbkPf0nYAbKSSUqKZIB59L07Gg6kk8ZEpkZEsyUOII0f1gQTZV44LtI0Ug8gMimJ5KzrlJCVDMs/tVj3RJ5bDmt+MI/kepZ/GmF8wwmcwKP6A6RfZZC32hH+Uo6mF2uNUktNbR0O8vVWjWL5Bml+0xRNFPENyhWOrgVOjeDuLXEsZwuRwVNRwBj2qIHG0RiIi9SxpRVLtZln2BxwA1BTK9duN5Z5hLdDRvhm8R+IPjqKK+42T3yO5Qe2ELnlAMufM74hpKUfgOAA42hMHFuiG7cOIh+zWEccH0bCoxMunW5cn+0EXtfwYq1RZFrfWOYR3kFnL7OiKzLuw+uudkAsMIKEb4Kv0iaQCY8eIPrAo5izGRgRRUT03aa2WzDtzBsfm/Ao1B+0u9Y67oZvrrB70QhW9RVX12IwGIhP/h2ltk4OPjXBb7kF5BaXycFOjbCZt3Q7O7v6EUoO3DWeeUov03ltsJnAQ8OBu2qZ0jdaw5Tepl4K4ips6llu+GhODto1lnmEd5QRXhNsK/Dmj+ZkSt9oNXXmKb1NvRTEVdjUs+xoLyN8UykypdcUmznxBho+mpOP5Rqntq6uYR9eY1N6g8FAful5enbrwi/HczhfXdPWJbWIzo4O/KF/w6cvthV40whfLSN8o9TU1mEwGD+V1dqUPqvgHCl5xfTspq3v6rOtwOtkH74pLj6tVjsj/JlzFWQWGL/Q07WzE8P8erZxRa3Hpt7W6/fhFRnhG8PywhlHjYzwlTUKOcVl1NTW4dmtCz5uXdu6pFZlU8+y5VF6OXB3fTUavFIup7iMwrLzdHLU4ePugk4jb3RmNvXXWh5plpNvrq/+whmLsxRtWX7peXKLjD+o4ePmgrtzpzauqPXZ1LPsaG+PnZ0dIPvxjWGeBTnYd7D5ka5KqSW3uJQqpZaeLl3w1thU3symnmWdfQeL8+llSn89WjrLLreolPzS83R00OHt3hVHDR2ktGRTgQfLL8KQEf56tHIefWHZeXKKywDwcetKdw1fSWlzz7ScT994Wjittqa2lpziMiprFLp37azZqbyZzT3T5qPN1bIPf11auDQ2p6iMM+cqcNTZ4+3WFScHmzr1pMlsLvCOOvksvrFsfUpfVH6BXNNU3tutq+bOqrsSm3um5Yq5xrPlg3ZKbR25xWWcr67B3bkTPm4ubV1Su2CDgZd9+May/FjO1uQWl3G6pBydfQd83Fzo5KjtqbyZzT3TjnLFXKMpFl9gaUvOVlSSU2w+waYrnq4ylTezucDLFXONYzz92PaO0tfVGcgpLqWiqga3Lk54y1T+IrbzTJt06egAQG2dTXwZb4vRdbCr/7/57ERbkFNcyqmz5dh3sMPbzaX+9SCMWnTHJreogo27s/j+8Cky8sooLKukpKK6JVcp2inXLo64dulIv55d+cOAnswM7UOAp7qjb8n5Souj8i70uslZ1f5tQYt8L71SW8eybcks2fIb00J8mRbiS6BXNzxdO+PaxVHt1QkrUFJRTWFZJWmnzvHdryf5PCGTR8b24/WZw3ByuPFjCAYDHD1p/I59185O3OLtQVcnea1dSvXAHz9ZwuzoBDo56lj1l7H06antM5vElRWWVvLMJwkczSkhJiIUfWD3G+ovu6iU5JOF2NnBLb27a/6MuqtRNfDHT5YwZtE2Fs4YwkvTBqvVrbBhX+zK5LlVu1n7/DjGDfRsVh/nLlRxNLeQ0gtV+Li7cEtvD5WrtB2q7cMrtXXMjk6QsIsmuf+2vjg52jM7OoFDS6bi7NT0g2y5xWWUXqjCpVNHzX2DTVOpdpR+2bZkOjnqJOyiyaaM8GPcQE9eWX+gycvmFpeRU9TwmbtLp45ql2dTVAl8blEFS7b8xqq/jFWjO6FB7z+mZ/PebPZnFjV6mbLK6voTbLzduspn7o2gSuA37s5iWoivHKATzXaTc0eemnAz63ZmNnqZnKJSzp2voquTI95uXbGh0wlajCqB//7wKaaF+KrRldCw6SP9+O7QyUa1PXm27KIr4Vw7O7VkaTZDlcBn5JUR6NVNja7alqGEz6NicHp9D/nm+2pSiHrq4vsMBUmEhf+L2d9X3Pg6lSw+eD4G5w9/u/G+rFxgr26k55Vet115VQ25xWXUGQz0uslZpvJNoErgC8sq8XS1ga8NsuvKoL7OVKcVc8z0y0O1qXlsLYaq40UcMd2npJTwk86NSYObcVFGXT7fffgFAR9LwC/l5GB/0XflX01uUSlnKyrp0tEBbzcX7DvIXL6xVAl8SUW1jZxBZ0/QAHf6VJew/zhAHaeSCzkc4MrtShG7fgWo5Xj6Wcr8PBjenI+N685zLKOUPEXVwjXjdEl5/ffTebu54NZFpvJNYXMXz9wo3RBPZjid5/vUPKCKI5mljBnWnz8FVPNT5kkwlJGcVYFXgAd9gbozx/jwzXU4h8fQ4dH1zP4yi0pAObiDyc+upmN4DB0e/IzxMcc4q2Txwbx4XjgJFb8kYffwt/xsWq+huogVC4ztb5q/nUTTweqq9P3MfdF4v9NjsSxOKEahluTPPsfuwc3MeX8dzhb92LLzVTXkFJdRW1eHp6uzfObeDBL4Szn2445bYEd2Pkp1Bj8edeD2fjczyL8jCcdOkl+TxU/HdNzfvx/UnuLTD/awNfA2CmOfoHReTw6vT+IfydDBrTd/+z/3UxH7BOee8uDQ98lsyfVn7tLxvN8buvxBj2HdJMaZVluZXEaPx8M580o/BuRk8fpPeXD+N/66+Dj2U6ZQEftnMh7oyHvRifxg3s2tLaXD4EmUWPRjy3KKyyguv2D81Ri3rjb/XfotQb4G5DIdGdS3G5XfFbD3VwOb6jxYNUTH0EoPnHcUc+RoRxLt3HlriI7a9HT+maaQmBZPp1jz8h0YVwS4V7N//bc8eew8v5dUUUw3lGtM4zsN9ef+oC6g+POAZwbvF5+jOjGbD0srqV7xJe+tMDW078bJfPAFcOhO2Hg3TTyJeecqyC3W9q/GqEELr5Um6kCvgR4ExxawMaGO07cGEOwAuptdGVedyaYfHEn282SAC3AasOvIGy8/yqIRFl3UneHfrybyutsQkj4Ipmfmz4x6o6BJVRRdMB0h1Hnwnw+mcf9FxwtqSd51Y3+lNblQrZBbXEpNbR09u2n3V2PUIHOiK7AP8mRyt3Ms3VnO6L496QrYufszud95/nOkgoEDvegL2Pfz4l73Kt7ZuIMjpQqGijx+iklgZ90FisoMuLk442ZfxsHdeeyr79wVr5ugtvQC+TXXrsNhcHdmGIpY8PlBTl2AmrNZrF+9l8afmmIbcotLKSy7gJODDm837f5qjBok8Ffi0I87b9VBh85MHtDbeF8Hdwb5O1F4oQNht/Q13qcLZN78oTxZlcmIJ1Zj/9QPLC7rRC+dHw/d60Xl9h10fmIrXzi5UT8BsOvG7aFe9DxyGM8nvmfnNcqw6z6CZS/2I+TYIXr/KYauL+xmo86FG7uQ1Lrkl1ZYHJXX9q/GqEGVy2N1Mz/FEDtbjXqExtmFx6B8/hgAVTW1/HaygILS83Tv2plbvT3oqPEfkrhRMsKLdiunuJSC0vM46uzxce8qYVeBBF60SwVl5y86V76Hi3zVtBok8KLdqVZqyS0y/gCk/GqMuiTwot3JLS7jTGmF8Vdj3OVXY9QkgRftTo7FCTae8gOQqpLAi3bnQrWCg30H/D1kKq821eZKduExanUlNM7NuRO9XZ3lqHwLUG2LyufwQg124TGM7OvV1mXYLJnSC6EhEnghNEQCL4SGSOCF0BAJvBAaIoEXQkMk8EJoiAReCA2xosCnsvaF+axNbes6Gi913XxeWGdFBQubZ0WBF0LcKBVPVk5l7QuL2JRjujn6GTbOu4PtS2fytdebvP9wkPH+7e8RdnAYG+fdQeq6+SwwLeA7Yzq9N50mOPZF7rxW3wtnssnUN6nreWHhV2QDoOfZ2Be5k1TWvvAJTBnOvo+/IhsfZix+F6+vZ7Ii0bKdcQSOZjIj9q401d3wGKbHzfVhuc5oGOH9FZsSje1Zau4b8JnO2+8/BPXLLiJsk55nY+/l9AufQMS7PBJk/pvMt3/k3fCDeM04xaZNMGOx+b6VGLv1Md2n3rMlNEpRFMON/iPsn4aLpRjWzJ1nWJNiMBhS1hnmzl1nSDE9Ev/uA4Yl8QaDIX6pYUbYUkO8eYm18y66fTmLPg2X92uIX2qY8e52U7sHDDPMj8UvNcwIM63TtP65a1Ms1tnQZ8raeaY+jP83t7uo7pR1hrkW/V2qvt1lfVxS/0W3txuWhD1g0Xa7YYnltrj0b7VhhP3zhl+P8u/q/9S9HGn7e4R9nFR/c3QucOdwRvAJe1MfIijoRw7kTmfqPEhddwrfGU/Vj6ZBD09m9KaDjV9X7mmyc5JYEP5Vw30+XqTSG/BhRsRDBAF4e+HrM52pphV5e/lc1I3vjKfqR86ghyczOvwg27kDTueQnbiIsE0Wbb1SwRuw6A+4ZKZhakdTh2M9U8yzoNQ8TpJEYvhMVlg8ntuMXoWwpF7gt79H2NdevB37OUHA9qUzOQBAECEhEL0vlZB9WzkZ8pTx8dM59B52gy9f8zT7IqnsvbFeG7r/y+fMv3T/InX/JbfX88LC00yJ/dxiN6QxEVsAAAPsSURBVEEFpl0DCbhQk2oH7VLzToG3p+kF+iMHEhseC3p4Mr33/pcte3vVj2J3DtOT+PV6zMewU9dtJZEm8PbCN3HrDR+1z967/+IaRg/jTowzAcv6rir3NNk+Xngbe2Dv3pyrNAzCyzuHfftMPabuZ99Vm3rSO+crtmxv0p8ixHWpNsIbp8MrCQtfCegZPdry0TsI9l7JCp5hvvmuO1/k2YMz66fkvjOmM5rT11oDISGwoP6g3UO8/5fThC2ciXnW7TvjTd5/uGl1+3qfJjp8pnE67jOdt9+/w/T3vMuzp2da7DKYDpxd2sGd9zLj60Wmdj6MHt2wyxA0YjgsNB+0e5E7p0zn64Wm3QQfPaN9Lu3M7A7mL87jhYUzCfvYdNcVZzNCNE37+SEK09HviFacxhqn3081fIIg2pzlD1EI9bWb7xDa/vVXEPImQVh+HGV28cdlQojmabvAX3Jkm9HPsPHhICCI+bEydRWiJbSfKb0QyJS+pcmptUJoiGpT+gMnitTqSmhMcB/3ti5BM1QLvDxpQrR/MqUXQkMk8EJoiAReCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggNkcALoSESeCE0RAIvhIZI4IXQEAm8EBoigRdCQyTwQmiIBF4IDZHAC6EhEnghNEQCL4SGSOCF0BAJvBAaIoEXQkMk8EJoiAReCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggNkcALoSESeCE0RAIvhIZI4IXQEAm8EBqiSuCdnRwor6xRoyshRAtSJfAeLk7klVxQoyuhYUptXVuXYPNUCXy/nl1JO3VOja6EhqWdPkeAp0tbl2HTVAn8Hwb05LtfT6rRldCwbw/mMu4Wz7Yuw6apEviZoX34PCGTgnMyrRfNU6PUErM9lYfH9G3rUmyaKoEP8HThkbH9iPjHLjW6Exr0t42H6N/blXEDZYRvSap9LPf6zGEczSnhi12ZanUpNOJAZgGfxKWwYvaoti7F5qkWeCcHe2IiQnlu1W5+/O2UWt0KG3cgs4BHlv3Cssf1eLg4tXU5Ns9OURSDmh3+nJzH7OgEJgd787+PhuDs5KBm98JGKLV1vBl7kE/iUlj2uJ6wUf5tXZImqB54gPLKGl5Zf4CtB3J54o5A7tP7E+DlgpODvdqrElZEqa0jPa+Ubw7kELM9lf69XVkxe5SM7K2oRQJvtj+ziHU7M/nu0EnS80rlxApBgKcL427x5OExfeUAXRto0cALIdoXuXhGCA2RwAuhIRJ4ITREAi+EhkjghdAQCbwQGiKBF0JDJPBCaIgEXggN+f/sQGmHiJF2nwAAAABJRU5ErkJggg==)\n","\n","The procedure to create the cube is the following.\n","\n","1. On the stations file:\n"," 1. replace empty states and countries in stations with a placeholder value (e.g., \"XX\");\n"," 1. keep only the following fields: stationId, state, country, elevation\n","2. On the weather-sample file:\n"," 1. filter out weather wrong measurements (i.e., where airTemperatureQuality=9);\n"," 1. keep only the following fields: stationId, airTemperature, date, month, year\n"," 1. create a new fulldate field by concatenating year, month, and date\n"," 1. create a new fullmonth field by concatenating year and month\n","1. Join stations with weather measurements on the stationId field\n","1. Keep only the following fields: state, country, elevation, fulldate, fullmonth, year, airTemperature\n","1. Aggregate the measurements by state and date to take the average temperature\n"," - Group by: state, country, elevation, fulldate, fullmonth, year\n"," - Calculation: avg(airTemperature)\n","1. Save the result on a file"],"metadata":{"id":"Dv5BUnaAWWEi"}},{"cell_type":"code","source":["!wget https://raw.githubusercontent.com/w4bo/2024-bbs-dm/master/materials/datasets/weather-stations.csv\n","!wget https://raw.githubusercontent.com/w4bo/2024-bbs-dm/master/materials/datasets/weather-sample-10k.txt"],"metadata":{"id":"REn3hjicbRFL"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["## Spark"],"metadata":{"id":"xO6bQ3NQwDAj"}},{"cell_type":"code","source":["dfW = sc.textFile(\"weather-sample-10k.txt\")\\\n"," .map(lambda l: (l[4:15],l[15:19],l[19:21],l[21:23],int(l[87:92])/10,l[92:93]))\\\n"," .toDF([\"stationId\",\"year\",\"month\",\"day\",\"airTemperature\",\"airTemperatureQuality\"])\n","dfW.show()"],"metadata":{"id":"Xcd4fAhgeDhH"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":["from pyspark.sql.functions import concat\n","dfS = spark.read.option(\"delimiter\", \",\").option(\"header\", \"false\").csv(\"weather-stations.csv\")\n","dfS = dfS.select(concat(dfS[0],dfS[1]),dfS[2],dfS[3],dfS[4],dfS[5],dfS[6],dfS[7],dfS[8],dfS[9],dfS[10])\\\n"," .toDF(\"stationId\",\"city\",\"country\",\"state\",\"call\",\"latitude\",\"longitude\",\"elevation\",\"date_begin\",\"date_end\")\n","dfS.show()"],"metadata":{"id":"OaknyiXHeCbF"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["1. On the stations file:\n"," 1. replace empty states and countries in stations with a placeholder value (e.g., \"XX\");\n"," 1. keep only the following fields: stationId, state, country, elevation"],"metadata":{"id":"8QqJZrglm-ZD"}},{"cell_type":"code","source":["dfS1 = dfS.fillna({'state': 'XX', 'country':'XX'})\n","dfS2 = dfS1.select('stationId','state','country','elevation')\n","dfS2.show()"],"metadata":{"id":"8v1YuN-4nQCQ"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["2. On the weather-sample file:\n"," 1. filter out weather wrong measurements (i.e., where airTemperatureQuality=9);\n"," 1. keep only the following fields: stationId, airTemperature, date, month, year\n"," 1. create a new fulldate field by concatenating year, month, and date\n"," 1. create a new fullmonth field by concatenating year and month"],"metadata":{"id":"5LxL1WQvnF0J"}},{"cell_type":"code","source":["from pyspark.sql.functions import concat, lit\n","dfW1 = dfW.where(\"airTemperature < 9\")\n","dfW2 = dfW1.select('stationId','airTemperature','day','month','year')\n","dfW3 = dfW2.withColumn(\"fulldate\", concat(dfW1.year,lit(\"-\"),dfW1.month,lit(\"-\"),dfW1.day))\n","dfW4 = dfW3.withColumn(\"fullmonth\", concat(dfW1.year,lit(\"-\"),dfW1.month))\n","dfW4.show()"],"metadata":{"id":"Cf9bm6CspR8x"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["3. Join stations with weather measurements on the stationId field\n"],"metadata":{"id":"jy2yp35unHhO"}},{"cell_type":"code","source":["dfJ = dfS2.join(dfW4, \"stationId\")\n","dfJ.show()"],"metadata":{"id":"oHTKQhQJrPMU"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["4. Keep only the following fields: state, country, elevation, fulldate, fullmonth, year, airTemperature"],"metadata":{"id":"mv_dfniznJ2A"}},{"cell_type":"code","source":["dfJ2 = dfJ.select(\"state\", \"country\", \"elevation\", \"fulldate\", \"fullmonth\", \"year\", \"airTemperature\")\n","dfJ2.show()"],"metadata":{"id":"dYyMxZaSrprK"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["5. Aggregate the measurements by state, country and date to take the average temperature\n"," - Group by: state, country, elevation, fulldate, fullmonth, year\n"," - Calculation: avg(airTemperature)"],"metadata":{"id":"YXKChhXwnLlv"}},{"cell_type":"code","source":["dfG = dfJ2.groupBy(\"state\", \"country\", \"elevation\", \"fulldate\", \"fullmonth\", \"year\").agg({'airTemperature': 'avg'})\n","dfG.show()"],"metadata":{"id":"shVr9k9hryqH"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["6. Save the result on a file"],"metadata":{"id":"zwXx8h-tv4-v"}},{"cell_type":"code","source":["dfG.write.mode('overwrite').option('header','true').csv(\"weather-cube\")"],"metadata":{"id":"pKTH93LQu61U"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["## PowerBI\n","\n","Download the file from the left panel of this notebook (in case of issues, download it from [here](https://raw.githubusercontent.com/w4bo/2024-bbs-dm/master/materials/results/weather-cube.csv)) and load it in [Power BI](https://app.powerbi.com/).\n","- Visualize the daily trend of average temperatures for each country\n","- Show the average temperature on the map\n","- Compute bins for the elevation field and show the average temperature for each bin\n","\n","\n","The final Power BI file will be available [here](https://raw.githubusercontent.com/w4bo/2024-bbs-dm/master/materials/results/weather-cube.pbix)."],"metadata":{"id":"jkKWX-mWv8S9"}},{"cell_type":"markdown","source":["# Additional exercises\n","\n","The solution will be available [here](https://raw.githubusercontent.com/w4bo/2024-bbs-dm/master/materials/results/2024_bbs_dm_spark_basics_solution.ipynb).\n","\n","## Getting familiar with data frame transformations\n","\n","Carry out the following operations (in any order).\n","\n","- ```.select()``` operator; starting from ```dfS```:\n"," 1. keep only country, elevation, date_begin, and date_end\n"," 1. keep only the first four characters of date_begin using [sf.substring](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.substring.html)\n"," 1. use [concat](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.concat.html) to concatenate date_begin with date_end and putting an underscore (\\_) in the middle; since the underscore is not a column, declare it as ```sf.lit(\"_\")```\n"," 1. use [coalesce](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.coalesce.html) to take the value of country if not null, otherwise the value of city\n"," 1. put countries in lowercase using [sf.lower](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.lower.html)\n"," 1. in the previous four points, use ```.alias()``` to give a meaningful name to the obtained columns\n","- ```.withColumn()``` operator; starting from ```dfS```:\n"," 1. do the same as points 2 to 5 of the ```select``` operator, but using ```withColumn```\n","- ```.filter()``` operator; starting from ```dfS```, keep only the rows where:\n"," 1. elevation is greater than 5000\n"," 1. country is not null, using [sf.isnotnull](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.isnotnull.html)\n"," 1. conditions 1 and 2 are both true; conditions must be put between parenthesis and separated by \"&\" (e.g., check [here](https://www.geeksforgeeks.org/pyspark-filter-dataframe-based-on-multiple-conditions/))\n"," 1. either one of conditions 1 and 2 is true; conditions must be put between parenthesis and separated by \"|\" (e.g., check [here](https://www.geeksforgeeks.org/pyspark-filter-dataframe-based-on-multiple-conditions/))\n"," 1. date_begin is the first day of the month (requires to use substring)\n","- ```.groupBy()``` operator; starting from ```dfW```:\n"," 1. group by airTemperatureQuality to count how many rows there are for each value\n"," 1. as above, but also calculate the average temperature\n"," 1. as above, but also given meaningful names to the results using ```withColumnRenamed```\n"," 1. group by month to calculate the minimum and maximum temperatures and order by month using ```orderby```; to aggregate differently on the same column, use the [sf.max](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.max.html) and [sf.min](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.min.html) functions inside the ```agg``` function instead of the object enclosed by brackets ```{}```\n"," 1. group by month and day to calculate the minimum and maximum temperatures\n"," 1. group by stationId and year month to count the number of rows"],"metadata":{"id":"679cmN5vLOMH"}},{"cell_type":"code","source":["import pyspark.sql.functions as sf\n","\n","# TODO HERE"],"metadata":{"id":"82YodmWqG-JZ"},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":["## Complete exercises\n","\n","Carry out the following exercises (in any order).\n","\n","- Check if there exist stations with a negative elevation; then, calculate how many of these stations exist in each country; rename the result to \"cnt\" and order the result by decreasing cnt\n","- Take only stations with positive elevation, compute the maximum elevation by country and rename the result to \"elevation\"; then, join the result with the original dfS to get, for each country, the name of the city with the highest elevation (join key: ```[\"country\",\"elevation\"]```); order the result by decreasing elevation\n","- Take only weather values with airQuality==1, compute the minimum temperature for each stationId and rename it to \"minTemperature\"; then, join the result with dfS and keep only the columns \"minTemperature\" and \"elevation\"; finally, use the correlation between the two columns. To do the last part, you need to:\n"," - cast the elevation to an integer datatype: you need to add ```from pyspark.sql.types import IntegerType``` and then ```df.myfield.cast(IntegerType())```;\n"," - compute the correlation with ```df.stat.corr(\"myfield1\",\"myfield2\")```."],"metadata":{"id":"JFGe5AvcHCKM"}},{"cell_type":"code","source":["# TODO HERE"],"metadata":{"id":"AG9YwgeaL_Db"},"execution_count":null,"outputs":[]}]} \ No newline at end of file