{ "cells": [ { "cell_type": "markdown", "id": "73e5b932-3752-40eb-a9c5-ccc37be9b779", "metadata": {}, "source": [ "## Content\n", "\n", "### Basic work with DataFrames\n", "- Creating a DataFrame from an RDD\n", "- Schema inference vs. explicitly defined schema\n", "- Default column names vs. explicitly defined column names\n", "- Creating a DataFrame from a CSV file\n", "\n", "### DataFrame API & SparkSQL\n", "- Actions: `show()`\n", "- Transformations: `filter()`, `select()`, `join()`, `groupBy()`\n", "- Aggregations: `agg()` and aggregate functions `avg()`, `max()`, `count()`\n", "- User-defined functions (UDFs)\n", "- SparkSQL\n", "- Displaying physical plan\n", "\n", "DataFrame API documentation\n", "\n", "\n" ] }, { "cell_type": "code", "execution_count": 1, "id": "297eb600-bfbf-4673-85c1-eae9f6e25848", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "WARNING: Using incubator modules: jdk.incubator.vector\n", "WARNING: package sun.security.action not in java.base\n", "Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties\n", "Setting default log level to \"WARN\".\n", "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n", "26/11/12 10:51:10 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n", "26/11/12 10:51:11 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n" ] } ], "source": [ "from pyspark.sql import SparkSession\n", "import os\n", "\n", "username = os.environ.get(\"JUPYTERHUB_USER\", \"unknown\")\n", "\n", "spark = (\n", " SparkSession.builder\n", " .master(\"spark://192.168.70.21:7077\")\n", " .appName(f\"Spark_Lab1_{username}\")\n", " .config(\"spark.driver.host\", \"192.168.70.21\")\n", " .config(\"spark.driver.bindAddress\", \"192.168.70.21\") \n", " .config(\"spark.cores.max\", \"2\")\n", " .config(\"spark.pyspark.python\", \"/opt/jupyterhub/venv/bin/python\") \n", " .config(\n", " \"spark.executorEnv.PYSPARK_PYTHON\",\n", " \"/opt/jupyterhub/venv/bin/python\"\n", " )\n", " .getOrCreate()\n", ")\n", "\n", "spark.sparkContext.setLogLevel(\"ERROR\")\n", "sc = spark.sparkContext" ] }, { "cell_type": "code", "execution_count": 2, "id": "bf086635-a5b1-49f8-b05f-2be570f8160e", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "data": { "text/plain": [ "[('', 268237),\n", " ('ET', 179),\n", " ('cupiat', 22),\n", " ('Haec', 178),\n", " ('legat', 53),\n", " ('quae', 16503),\n", " ('liber', 389),\n", " ('tenet', 392),\n", " ('Scripserat', 4),\n", " ('haec', 6804),\n", " ('inter', 5539),\n", " ('rex', 2823),\n", " ('Karulus', 2),\n", " ('Albinusque', 1),\n", " ('hic', 4063),\n", " ('probat', 84),\n", " ('opus', 1493),\n", " ('amborum', 44),\n", " ('dispar', 20),\n", " ('sed', 15031)]" ] }, "execution_count": 2, "metadata": {}, "output_type": "execute_result" } ], "source": [ "latinlinesRDD = sc.textFile(\"/datasets/latincorpus.txt\")\n", "\n", "rawlatinwordsRDD = latinlinesRDD.flatMap(lambda line: line.split(\" \"))\n", "\n", "latinwordsRDD = rawlatinwordsRDD.map(\n", " lambda word: ''.join(c for c in word if c not in \".,:; '\")\n", ")\n", "\n", "wordcountRDD = latinwordsRDD.map(lambda word: (word, 1)) \\\n", " .reduceByKey(lambda a, b: a + b)\n", "\n", "wordcountRDD.take(20)" ] }, { "cell_type": "code", "execution_count": 3, "id": "08050d75-f8e1-4750-8941-4ab0dd9ee36f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- _1: string (nullable = true)\n", " |-- _2: long (nullable = true)\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 5:> (0 + 1) / 1]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-----------+------+\n", "| _1| _2|\n", "+-----------+------+\n", "| |268237|\n", "| Theodonis| 8|\n", "| cum| 29505|\n", "| Hluthario| 2|\n", "| et|140057|\n", "|supradictus| 48|\n", "| Bernhardus| 13|\n", "|purificavit| 5|\n", "| supradicto| 19|\n", "| stupro| 39|\n", "| postquam| 1182|\n", "| inventus| 129|\n", "| est| 41705|\n", "| qui| 26221|\n", "| ausus| 292|\n", "| armis| 1208|\n", "| quod| 23188|\n", "| eius| 10108|\n", "| Hlutharii| 3|\n", "| patrem| 815|\n", "+-----------+------+\n", "only showing top 20 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "wordcountDF = wordcountRDD.toDF()\n", "\n", "wordcountDF.printSchema()\n", "wordcountDF.show()" ] }, { "cell_type": "code", "execution_count": 4, "id": "007ab9e9-c2a4-43f7-97b0-bfc71331838d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- word: string (nullable = true)\n", " |-- count: long (nullable = true)\n", "\n", "+-----------+------+\n", "| word| count|\n", "+-----------+------+\n", "| |268237|\n", "| Theodonis| 8|\n", "| cum| 29505|\n", "| Hluthario| 2|\n", "| et|140057|\n", "|supradictus| 48|\n", "| Bernhardus| 13|\n", "|purificavit| 5|\n", "| supradicto| 19|\n", "| stupro| 39|\n", "| postquam| 1182|\n", "| inventus| 129|\n", "| est| 41705|\n", "| qui| 26221|\n", "| ausus| 292|\n", "| armis| 1208|\n", "| quod| 23188|\n", "| eius| 10108|\n", "| Hlutharii| 3|\n", "| patrem| 815|\n", "+-----------+------+\n", "only showing top 20 rows\n" ] } ], "source": [ "wordcountDF = wordcountRDD.toDF([\"word\", \"count\"])\n", "\n", "wordcountDF.printSchema()\n", "wordcountDF.show()" ] }, { "cell_type": "code", "execution_count": 5, "id": "ba19f054-638f-43a9-9c82-73d82b3022bc", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- word: string (nullable = false)\n", " |-- count: integer (nullable = true)\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 11:> (0 + 1) / 1]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+----------+------+\n", "| word| count|\n", "+----------+------+\n", "| |268237|\n", "| ET| 179|\n", "| cupiat| 22|\n", "| Haec| 178|\n", "| legat| 53|\n", "| quae| 16503|\n", "| liber| 389|\n", "| tenet| 392|\n", "|Scripserat| 4|\n", "| haec| 6804|\n", "| inter| 5539|\n", "| rex| 2823|\n", "| Karulus| 2|\n", "|Albinusque| 1|\n", "| hic| 4063|\n", "| probat| 84|\n", "| opus| 1493|\n", "| amborum| 44|\n", "| dispar| 20|\n", "| sed| 15031|\n", "+----------+------+\n", "only showing top 20 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql.types import StructType, StructField, StringType, IntegerType\n", "\n", "column_schema = StructType([\n", " StructField(\"word\", StringType(), False),\n", " StructField(\"count\", IntegerType(), True)\n", "])\n", "\n", "wordcountDF = spark.createDataFrame(wordcountRDD, column_schema)\n", "wordcountDF.printSchema()\n", "wordcountDF.show()" ] }, { "cell_type": "code", "execution_count": 6, "id": "f3a95050-5ac4-453f-acb8-9c57ee2643b2", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "# DataFrame is a collection of Row objects organised according to a schema\n", "print(type(wordcountDF))\n", "\n", "row = wordcountDF.first()\n", "print(type(row))\n", "\n" ] }, { "cell_type": "code", "execution_count": 7, "id": "59ffdd56-453b-489b-b955-bf09d80e47b9", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- word: string (nullable = false)\n", " |-- count: integer (nullable = true)\n", "\n", "+----------+------+\n", "| word| count|\n", "+----------+------+\n", "| |268237|\n", "| ET| 179|\n", "| cupiat| 22|\n", "| Haec| 178|\n", "| legat| 53|\n", "| quae| 16503|\n", "| liber| 389|\n", "| tenet| 392|\n", "|Scripserat| 4|\n", "| haec| 6804|\n", "| inter| 5539|\n", "| rex| 2823|\n", "| Karulus| 2|\n", "|Albinusque| 1|\n", "| hic| 4063|\n", "| probat| 84|\n", "| opus| 1493|\n", "| amborum| 44|\n", "| dispar| 20|\n", "| sed| 15031|\n", "+----------+------+\n", "only showing top 20 rows\n" ] } ], "source": [ "from pyspark.sql import Row\n", "from pyspark.sql.types import (\n", " StructType, StructField, StringType, IntegerType\n", ")\n", "\n", "column_schema = StructType([\n", " StructField(\"word\", StringType(), False),\n", " StructField(\"count\", IntegerType(), True)\n", "])\n", "\n", "# Row is also a constructor - each tuple is converted to a Row object\n", "wordcountRowRDD = wordcountRDD.map(\n", " lambda pair: Row(pair[0], pair[1])\n", ")\n", "\n", "wordcountRowRDD.take(10)\n", "\n", "wordcountDF = spark.createDataFrame(wordcountRowRDD, column_schema)\n", "\n", "wordcountDF.printSchema()\n", "wordcountDF.show()" ] }, { "cell_type": "code", "execution_count": 8, "id": "f3868ccd-a54c-4ad2-a6d7-9bc39549fd3f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+------+\n", "| word| count|\n", "+-------+------+\n", "| |268237|\n", "| ET| 179|\n", "| cupiat| 22|\n", "| Haec| 178|\n", "| legat| 53|\n", "| quae| 16503|\n", "| liber| 389|\n", "| tenet| 392|\n", "| haec| 6804|\n", "| inter| 5539|\n", "| rex| 2823|\n", "| hic| 4063|\n", "| probat| 84|\n", "| opus| 1493|\n", "|amborum| 44|\n", "| sed| 15031|\n", "| pater| 1888|\n", "| mundi| 541|\n", "| inops| 91|\n", "| modico| 87|\n", "+-------+------+\n", "only showing top 20 rows\n" ] } ], "source": [ "# DataFrame column API\n", "wordcountDF.filter(wordcountDF[\"count\"] > 20).show()\n", "\n", "# Alternative - SQL expression string\n", "# wordcountDF.filter(\"count > 20\").show()" ] }, { "cell_type": "code", "execution_count": 9, "id": "ff35576d-b5c1-4a6b-8ecb-653d65261cff", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 21:> (0 + 2) / 2]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+----+------+\n", "|word| count|\n", "+----+------+\n", "| |268237|\n", "| et|140057|\n", "| in| 80665|\n", "| est| 41705|\n", "| non| 38493|\n", "| ad| 36322|\n", "| ut| 34516|\n", "| cum| 29505|\n", "| qui| 26221|\n", "|quod| 23188|\n", "+----+------+\n", "only showing top 10 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "wordcountDF.filter(wordcountDF[\"count\"] > 20) \\\n", " .orderBy(wordcountDF[\"count\"].desc()) \\\n", " .show(10)\n", "\n", "# wordcountDF.filter(\"count > 20\") \\\n", "# .orderBy(\"count\", ascending=False) \\\n", "# .show(10)" ] }, { "cell_type": "code", "execution_count": 10, "id": "0a1afb82-e65e-41a1-898e-17442fbc30b3", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "keyCompany,stockCodeCompany,nameCompany,sectorCodeCompany,sectorCompany,segmentCompany\n", "1,BRAP4,BRADESPAR,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "2,PMAM3,PARANAPANEMA,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "3,CSNA3,SID NACIONAL,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "4,SUZB3,SUZANO S.A.,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "5,BRKM5,BRASKEM,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "6,RANI3,IRANI,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "7,DEXP3,DEXXOS PAR,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "8,GGBR4,GERDAU,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "9,UNIP6,UNIPAR,IMAT,BM&FBOVESPA BASIC MATERIALS INDEX (IMAT),SEGMENTS AND SECTORS\n", "keyTime,datetime,dayTime,dayWeekTime,dayWeekAbbrevTime,dayWeekCompleteTime,monthTime,monthAbbrevTime,monthCompleteTime,bimonthTime,quarterTime,semesterTime,yearTime\n", "1,1994-07-01,1,6,SAB,SABADO,7,JUL,JULHO,4,3,2,1994\n", "2,1994-07-02,2,7,DOM,DOMINGO,7,JUL,JULHO,4,3,2,1994\n", "3,1994-07-03,3,1,SEG,SEGUNDA,7,JUL,JULHO,4,3,2,1994\n", "4,1994-07-04,4,2,TER,TERCA,7,JUL,JULHO,4,3,2,1994\n", "5,1994-07-05,5,3,QUA,QUARTA,7,JUL,JULHO,4,3,2,1994\n", "6,1994-07-06,6,4,QUI,QUINTA,7,JUL,JULHO,4,3,2,1994\n", "7,1994-07-07,7,5,SEX,SEXTA,7,JUL,JULHO,4,3,2,1994\n", "8,1994-07-08,8,6,SAB,SABADO,7,JUL,JULHO,4,3,2,1994\n", "9,1994-07-09,9,7,DOM,DOMINGO,7,JUL,JULHO,4,3,2,1994\n", "keyTime,keyCompany,openValueStock,closeValueStock,highValueStock,lowValueStock,quantityStock\n", "1355,18,45.8,45.8,45.8,45.8,916.0\n", "1355,107,12.15,12.6,12.6,12.15,668420.0\n", "1355,108,17.0,17.5,17.5,16.95,3789172.0\n", "1355,604,299.0,299.0,299.0,299.0,59800.0\n", "1355,110,53.1,52.0,53.8,52.0,8331641.0\n", "1355,111,57.0,56.5,57.0,55.5,9327191.0\n", "1355,100,10.9,10.95,10.95,10.65,8684946.4\n", "1355,11,42.0,41.89,42.0,41.8,171670.0\n", "1355,573,40.0,41.0,41.0,40.0,10010.0\n" ] } ], "source": [ "!head /datasets/dimCompany.csv\n", "!head /datasets/dimTime.csv\n", "!head /datasets/factStocks.csv" ] }, { "cell_type": "code", "execution_count": 11, "id": "1f765a11-ec06-4e6b-97e4-2fd264de98c2", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- keyCompany: string (nullable = true)\n", " |-- stockCodeCompany: string (nullable = true)\n", " |-- nameCompany: string (nullable = true)\n", " |-- sectorCodeCompany: string (nullable = true)\n", " |-- sectorCompany: string (nullable = true)\n", " |-- segmentCompany: string (nullable = true)\n", "\n", "root\n", " |-- keyTime: string (nullable = true)\n", " |-- datetime: string (nullable = true)\n", " |-- dayTime: string (nullable = true)\n", " |-- dayWeekTime: string (nullable = true)\n", " |-- dayWeekAbbrevTime: string (nullable = true)\n", " |-- dayWeekCompleteTime: string (nullable = true)\n", " |-- monthTime: string (nullable = true)\n", " |-- monthAbbrevTime: string (nullable = true)\n", " |-- monthCompleteTime: string (nullable = true)\n", " |-- bimonthTime: string (nullable = true)\n", " |-- quarterTime: string (nullable = true)\n", " |-- semesterTime: string (nullable = true)\n", " |-- yearTime: string (nullable = true)\n", "\n", "root\n", " |-- keyTime: string (nullable = true)\n", " |-- keyCompany: string (nullable = true)\n", " |-- openValueStock: string (nullable = true)\n", " |-- closeValueStock: string (nullable = true)\n", " |-- highValueStock: string (nullable = true)\n", " |-- lowValueStock: string (nullable = true)\n", " |-- quantityStock: string (nullable = true)\n", "\n" ] } ], "source": [ "# create dataframes for each csv\n", "dimCompany = spark.read.csv(\"/datasets/dimCompany.csv\", sep=\",\", header=True)\n", "dimTime = spark.read.csv(\"/datasets/dimTime.csv\", sep=\",\", header=True)\n", "factStocks = spark.read.csv(\"/datasets/factStocks.csv\", sep=\",\", header=True)\n", "\n", "# schema is not inferred by default - all columns have string type\n", "dimCompany.printSchema()\n", "dimTime.printSchema()\n", "factStocks.printSchema()\n", "\n", "# you can set inferSchema=True to infer schema\n" ] }, { "cell_type": "code", "execution_count": null, "id": "135382f7-3706-4455-809a-4e4034034cd5", "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": 12, "id": "a00042e4-99e6-4366-82d8-c397cf4d7079", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- keyCompany: integer (nullable = true)\n", " |-- stockCodeCompany: string (nullable = true)\n", " |-- nameCompany: string (nullable = true)\n", " |-- sectorCodeCompany: string (nullable = true)\n", " |-- sectorCompany: string (nullable = true)\n", " |-- segmentCompany: string (nullable = true)\n", "\n", "+----------+----------------+------------+-----------------+--------------------+--------------------+\n", "|keyCompany|stockCodeCompany| nameCompany|sectorCodeCompany| sectorCompany| segmentCompany|\n", "+----------+----------------+------------+-----------------+--------------------+--------------------+\n", "| 1| BRAP4| BRADESPAR| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 2| PMAM3|PARANAPANEMA| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 3| CSNA3|SID NACIONAL| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 4| SUZB3| SUZANO S.A.| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 5| BRKM5| BRASKEM| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 6| RANI3| IRANI| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 7| DEXP3| DEXXOS PAR| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 8| GGBR4| GERDAU| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 9| UNIP6| UNIPAR| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 10| KLBN11| KLABIN S/A| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 11| GOAU4| GERDAU MET| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 12| DXCO3| DEXCO| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 13| USIM3| USIMINAS| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 14| USIM5| USIMINAS| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 15| FESA4| FERBASA| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 16| VALE3| VALE| IMAT|BM&FBOVESPA BASIC...|SEGMENTS AND SECTORS|\n", "| 17| AALR3| ALLIAR| ICON|BM&FBOVESPA CONSU...|SEGMENTS AND SECTORS|\n", "| 18| ALPA4| ALPARGATAS| ICON|BM&FBOVESPA CONSU...|SEGMENTS AND SECTORS|\n", "| 19| ABEV3| AMBEV S/A| ICON|BM&FBOVESPA CONSU...|SEGMENTS AND SECTORS|\n", "| 20| AMER3| AMERICANAS| ICON|BM&FBOVESPA CONSU...|SEGMENTS AND SECTORS|\n", "+----------+----------------+------------+-----------------+--------------------+--------------------+\n", "only showing top 20 rows\n" ] } ], "source": [ "from pyspark.sql.types import (\n", " StructType, StructField, IntegerType, StringType\n", ")\n", "\n", "dimCompany_schema = StructType() \\\n", " .add(\"keyCompany\", IntegerType(), False) \\\n", " .add(\"stockCodeCompany\", StringType(), True) \\\n", " .add(\"nameCompany\", StringType(), True) \\\n", " .add(\"sectorCodeCompany\", StringType(), True) \\\n", " .add(\"sectorCompany\", StringType(), True) \\\n", " .add(\"segmentCompany\", StringType(), True)\n", "\n", "dimCompany = spark.read \\\n", " .options(sep=\",\", header=\"true\") \\\n", " .schema(dimCompany_schema) \\\n", " .csv(\"/datasets/dimCompany.csv\")\n", "\n", "dimCompany.printSchema()\n", "dimCompany.show()" ] }, { "cell_type": "code", "execution_count": 13, "id": "0017e897-8be8-4303-8f6a-50ef9a3c660a", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- keyTime: integer (nullable = true)\n", " |-- datetime: string (nullable = true)\n", " |-- dayTime: integer (nullable = true)\n", " |-- dayWeekTime: integer (nullable = true)\n", " |-- dayWeekAbbrevTime: string (nullable = true)\n", " |-- dayWeekCompleteTime: string (nullable = true)\n", " |-- monthTime: integer (nullable = true)\n", " |-- monthAbbrevTime: string (nullable = true)\n", " |-- monthCompleteTime: string (nullable = true)\n", " |-- bimonthTime: integer (nullable = true)\n", " |-- quarterTime: integer (nullable = true)\n", " |-- semesterTime: integer (nullable = true)\n", " |-- yearTime: integer (nullable = true)\n", "\n", "+-------+----------+-------+-----------+-----------------+-------------------+---------+---------------+-----------------+-----------+-----------+------------+--------+\n", "|keyTime| datetime|dayTime|dayWeekTime|dayWeekAbbrevTime|dayWeekCompleteTime|monthTime|monthAbbrevTime|monthCompleteTime|bimonthTime|quarterTime|semesterTime|yearTime|\n", "+-------+----------+-------+-----------+-----------------+-------------------+---------+---------------+-----------------+-----------+-----------+------------+--------+\n", "| 1|1994-07-01| 1| 6| SAB| SABADO| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 2|1994-07-02| 2| 7| DOM| DOMINGO| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 3|1994-07-03| 3| 1| SEG| SEGUNDA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 4|1994-07-04| 4| 2| TER| TERCA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 5|1994-07-05| 5| 3| QUA| QUARTA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 6|1994-07-06| 6| 4| QUI| QUINTA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 7|1994-07-07| 7| 5| SEX| SEXTA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 8|1994-07-08| 8| 6| SAB| SABADO| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 9|1994-07-09| 9| 7| DOM| DOMINGO| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 10|1994-07-10| 10| 1| SEG| SEGUNDA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 11|1994-07-11| 11| 2| TER| TERCA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 12|1994-07-12| 12| 3| QUA| QUARTA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 13|1994-07-13| 13| 4| QUI| QUINTA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 14|1994-07-14| 14| 5| SEX| SEXTA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 15|1994-07-15| 15| 6| SAB| SABADO| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 16|1994-07-16| 16| 7| DOM| DOMINGO| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 17|1994-07-17| 17| 1| SEG| SEGUNDA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 18|1994-07-18| 18| 2| TER| TERCA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 19|1994-07-19| 19| 3| QUA| QUARTA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "| 20|1994-07-20| 20| 4| QUI| QUINTA| 7| JUL| JULHO| 4| 3| 2| 1994|\n", "+-------+----------+-------+-----------+-----------------+-------------------+---------+---------------+-----------------+-----------+-----------+------------+--------+\n", "only showing top 20 rows\n" ] } ], "source": [ "from pyspark.sql.types import StructType, IntegerType, StringType\n", "\n", "dimTime_schema = StructType() \\\n", " .add(\"keyTime\", IntegerType(), False) \\\n", " .add(\"datetime\", StringType(), True) \\\n", " .add(\"dayTime\", IntegerType(), True) \\\n", " .add(\"dayWeekTime\", IntegerType(), True) \\\n", " .add(\"dayWeekAbbrevTime\", StringType(), True) \\\n", " .add(\"dayWeekCompleteTime\", StringType(), True) \\\n", " .add(\"monthTime\", IntegerType(), True) \\\n", " .add(\"monthAbbrevTime\", StringType(), True) \\\n", " .add(\"monthCompleteTime\", StringType(), True) \\\n", " .add(\"bimonthTime\", IntegerType(), True) \\\n", " .add(\"quarterTime\", IntegerType(), True) \\\n", " .add(\"semesterTime\", IntegerType(), True) \\\n", " .add(\"yearTime\", IntegerType(), True)\n", "\n", "dimTime = spark.read \\\n", " .options(sep=\",\", header=\"true\") \\\n", " .schema(dimTime_schema) \\\n", " .csv(\"/datasets/dimTime.csv\")\n", "\n", "dimTime.printSchema()\n", "dimTime.show()" ] }, { "cell_type": "code", "execution_count": 14, "id": "e9b8e1ab-7654-48da-95c4-258048cc09a5", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- keyTime: integer (nullable = true)\n", " |-- keyCompany: integer (nullable = true)\n", " |-- openValueStock: float (nullable = true)\n", " |-- closeValueStock: float (nullable = true)\n", " |-- highValueStock: float (nullable = true)\n", " |-- lowValueStock: float (nullable = true)\n", " |-- quantityStock: float (nullable = true)\n", "\n", "+-------+----------+--------------+---------------+--------------+-------------+-------------+\n", "|keyTime|keyCompany|openValueStock|closeValueStock|highValueStock|lowValueStock|quantityStock|\n", "+-------+----------+--------------+---------------+--------------+-------------+-------------+\n", "| 1355| 18| 45.8| 45.8| 45.8| 45.8| 916.0|\n", "| 1355| 107| 12.15| 12.6| 12.6| 12.15| 668420.0|\n", "| 1355| 108| 17.0| 17.5| 17.5| 16.95| 3789172.0|\n", "| 1355| 604| 299.0| 299.0| 299.0| 299.0| 59800.0|\n", "| 1355| 110| 53.1| 52.0| 53.8| 52.0| 8331641.0|\n", "| 1355| 111| 57.0| 56.5| 57.0| 55.5| 9327191.0|\n", "| 1355| 100| 10.9| 10.95| 10.95| 10.65| 8684946.0|\n", "| 1355| 11| 42.0| 41.89| 42.0| 41.8| 171670.0|\n", "| 1355| 573| 40.0| 41.0| 41.0| 40.0| 10010.0|\n", "| 1355| 115| 19.89| 19.89| 19.89| 19.89| 25857.0|\n", "| 1355| 116| 0.94| 0.96| 0.97| 0.93| 650240.0|\n", "| 1355| 62| 6.0| 5.7| 6.0| 5.7| 422922.0|\n", "| 1355| 235| 218.0| 215.03| 218.0| 215.0| 724584.3|\n", "| 1355| 99| 10.5| 10.5| 10.5| 10.5| 613200.0|\n", "| 1355| 104| 52.3| 52.2| 52.5| 52.0| 6851345.0|\n", "| 1355| 580| 43.4| 41.5| 43.4| 41.5| 804216.0|\n", "| 1355| 101| 8.1| 8.0| 8.1| 7.9| 120403.0|\n", "| 1355| 103| 40.2| 39.5| 40.2| 39.5| 404603.0|\n", "| 1355| 601| 0.23| 0.23| 0.23| 0.23| 37720.0|\n", "| 1355| 119| 276.0| 275.0| 276.0| 273.99| 2.7078934E7|\n", "+-------+----------+--------------+---------------+--------------+-------------+-------------+\n", "only showing top 20 rows\n" ] } ], "source": [ "from pyspark.sql.types import StructType, IntegerType, FloatType\n", "\n", "factStocks_schema = StructType() \\\n", " .add(\"keyTime\", IntegerType(), False) \\\n", " .add(\"keyCompany\", IntegerType(), False) \\\n", " .add(\"openValueStock\", FloatType(), True) \\\n", " .add(\"closeValueStock\", FloatType(), True) \\\n", " .add(\"highValueStock\", FloatType(), True) \\\n", " .add(\"lowValueStock\", FloatType(), True) \\\n", " .add(\"quantityStock\", FloatType(), True)\n", "\n", "factStocks = spark.read \\\n", " .options(sep=\",\", header=\"true\") \\\n", " .schema(factStocks_schema) \\\n", " .csv(\"/datasets/factStocks.csv\")\n", "\n", "factStocks.printSchema()\n", "factStocks.show()" ] }, { "cell_type": "code", "execution_count": 15, "id": "7bf3c976-062b-45cc-aebc-78cdafbc4857", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+--------------+\n", "|keyTime|openValueStock|\n", "+-------+--------------+\n", "| 1355| 45.8|\n", "| 1355| 12.15|\n", "| 1355| 17.0|\n", "| 1355| 299.0|\n", "| 1355| 53.1|\n", "| 1355| 57.0|\n", "| 1355| 10.9|\n", "| 1355| 42.0|\n", "| 1355| 40.0|\n", "| 1355| 19.89|\n", "| 1355| 0.94|\n", "| 1355| 6.0|\n", "| 1355| 218.0|\n", "| 1355| 10.5|\n", "| 1355| 52.3|\n", "| 1355| 43.4|\n", "| 1355| 8.1|\n", "| 1355| 40.2|\n", "| 1355| 0.23|\n", "| 1355| 276.0|\n", "+-------+--------------+\n", "only showing top 20 rows\n", "+-------+--------------------+\n", "|keyTime|(openValueStock + 1)|\n", "+-------+--------------------+\n", "| 1355| 46.79999923706055|\n", "| 1355| 13.149999618530273|\n", "| 1355| 18.0|\n", "| 1355| 300.0|\n", "| 1355| 54.099998474121094|\n", "| 1355| 58.0|\n", "| 1355| 11.899999618530273|\n", "| 1355| 43.0|\n", "| 1355| 41.0|\n", "| 1355| 20.889999389648438|\n", "| 1355| 1.9399999976158142|\n", "| 1355| 7.0|\n", "| 1355| 219.0|\n", "| 1355| 11.5|\n", "| 1355| 53.29999923706055|\n", "| 1355| 44.400001525878906|\n", "| 1355| 9.100000381469727|\n", "| 1355| 41.20000076293945|\n", "| 1355| 1.2300000041723251|\n", "| 1355| 277.0|\n", "+-------+--------------------+\n", "only showing top 20 rows\n" ] } ], "source": [ "factStocks.select(\"keyTime\", \"openValueStock\").show()\n", "\n", "factStocks.select(factStocks[\"keyTime\"], factStocks[\"openValueStock\"] + 1).show()" ] }, { "cell_type": "code", "execution_count": 16, "id": "e9a76d7e-8034-479a-88c8-54d456687c83", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- result: double (nullable = true)\n", "\n", "+--------------+------------------------------------+\n", "|openValueStock|CAST((openValueStock + 1) AS STRING)|\n", "+--------------+------------------------------------+\n", "| 45.8| 46.79999923706055|\n", "| 12.15| 13.149999618530273|\n", "| 17.0| 18.0|\n", "| 299.0| 300.0|\n", "| 53.1| 54.099998474121094|\n", "| 57.0| 58.0|\n", "| 10.9| 11.899999618530273|\n", "| 42.0| 43.0|\n", "| 40.0| 41.0|\n", "| 19.89| 20.889999389648438|\n", "| 0.94| 1.9399999976158142|\n", "| 6.0| 7.0|\n", "| 218.0| 219.0|\n", "| 10.5| 11.5|\n", "| 52.3| 53.29999923706055|\n", "| 43.4| 44.400001525878906|\n", "| 8.1| 9.100000381469727|\n", "| 40.2| 41.20000076293945|\n", "| 0.23| 1.2300000041723251|\n", "| 276.0| 277.0|\n", "+--------------+------------------------------------+\n", "only showing top 20 rows\n" ] } ], "source": [ "# the result of arithmetic operation is treated as double\n", "factStocks.select((factStocks[\"openValueStock\"] + 1).alias(\"result\")).printSchema()\n", "\n", "# float-to-string cast and double-to-string cast works in a different way\n", "factStocks.selectExpr(\n", " \"CAST(openValueStock AS STRING)\",\n", " \"CAST(openValueStock + 1 AS STRING)\"\n", ").show()" ] }, { "cell_type": "code", "execution_count": 17, "id": "298eb965-cc59-4add-b83a-93de295ab526", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+----------+--------+---------------+-------+\n", "|keyTime| datetime|yearTime|monthAbbrevTime|dayTime|\n", "+-------+----------+--------+---------------+-------+\n", "| 1|1994-07-01| 1994| JUL| 1|\n", "| 2|1994-07-02| 1994| JUL| 2|\n", "| 3|1994-07-03| 1994| JUL| 3|\n", "| 4|1994-07-04| 1994| JUL| 4|\n", "| 5|1994-07-05| 1994| JUL| 5|\n", "| 6|1994-07-06| 1994| JUL| 6|\n", "| 7|1994-07-07| 1994| JUL| 7|\n", "| 8|1994-07-08| 1994| JUL| 8|\n", "| 9|1994-07-09| 1994| JUL| 9|\n", "| 10|1994-07-10| 1994| JUL| 10|\n", "| 11|1994-07-11| 1994| JUL| 11|\n", "| 12|1994-07-12| 1994| JUL| 12|\n", "| 13|1994-07-13| 1994| JUL| 13|\n", "| 14|1994-07-14| 1994| JUL| 14|\n", "| 15|1994-07-15| 1994| JUL| 15|\n", "| 16|1994-07-16| 1994| JUL| 16|\n", "| 17|1994-07-17| 1994| JUL| 17|\n", "| 18|1994-07-18| 1994| JUL| 18|\n", "| 19|1994-07-19| 1994| JUL| 19|\n", "| 20|1994-07-20| 1994| JUL| 20|\n", "+-------+----------+--------+---------------+-------+\n", "only showing top 20 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 32:> (0 + 1) / 1]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-------+----------+-----------+\n", "|keyTime| datetime| myDate|\n", "+-------+----------+-----------+\n", "| 1|1994-07-01| 1994.JUL.1|\n", "| 2|1994-07-02| 1994.JUL.2|\n", "| 3|1994-07-03| 1994.JUL.3|\n", "| 4|1994-07-04| 1994.JUL.4|\n", "| 5|1994-07-05| 1994.JUL.5|\n", "| 6|1994-07-06| 1994.JUL.6|\n", "| 7|1994-07-07| 1994.JUL.7|\n", "| 8|1994-07-08| 1994.JUL.8|\n", "| 9|1994-07-09| 1994.JUL.9|\n", "| 10|1994-07-10|1994.JUL.10|\n", "| 11|1994-07-11|1994.JUL.11|\n", "| 12|1994-07-12|1994.JUL.12|\n", "| 13|1994-07-13|1994.JUL.13|\n", "| 14|1994-07-14|1994.JUL.14|\n", "| 15|1994-07-15|1994.JUL.15|\n", "| 16|1994-07-16|1994.JUL.16|\n", "| 17|1994-07-17|1994.JUL.17|\n", "| 18|1994-07-18|1994.JUL.18|\n", "| 19|1994-07-19|1994.JUL.19|\n", "| 20|1994-07-20|1994.JUL.20|\n", "+-------+----------+-----------+\n", "only showing top 20 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql.functions import udf\n", "from pyspark.sql.types import StringType\n", "\n", "dimTime.select(\"keyTime\", \"datetime\", \"yearTime\", \"monthAbbrevTime\", \"dayTime\").show()\n", "\n", "# udf - user-defined function\n", "# - takes three parameters\n", "# - returns their concatenation, separating them with dots\n", "# - rrturn type = StringType\n", "column_def = udf(lambda col1, col2, col3: str(col1) + \".\" + str(col2) + \".\" + str(col3),StringType())\n", "\n", "dimTime.select(\"keyTime\", \"datetime\", column_def(\"yearTime\", \"monthAbbrevTime\", \"dayTime\").alias(\"myDate\")).show()" ] }, { "cell_type": "code", "execution_count": 18, "id": "4ef30c40-dbdc-47e4-a3d6-075e3acd5b7f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------+------------+--------------+\n", "|yearTime| nameCompany|highValueStock|\n", "+--------+------------+--------------+\n", "| 1998| ALPARGATAS| 45.8|\n", "| 1998| COPEL| 12.6|\n", "| 1998| COPEL| 17.5|\n", "| 1998| COTEMINAS| 299.0|\n", "| 1998| ELETROBRAS| 53.8|\n", "| 1998| ELETROBRAS| 57.0|\n", "| 1998| BRADESCO| 10.95|\n", "| 1998| GERDAU MET| 42.0|\n", "| 1998| BANRISUL| 41.0|\n", "| 1998| INDS ROMI| 19.89|\n", "| 1998| ITAUSA| 0.97|\n", "| 1998|LOJAS AMERIC| 6.0|\n", "| 1998| PETROBRAS| 218.0|\n", "| 1998| BRADESCO| 10.5|\n", "| 1998| CEMIG| 52.5|\n", "| 1998| CESP| 43.4|\n", "| 1998| BRASIL| 8.1|\n", "| 1998| CEMIG| 40.2|\n", "| 1998| BANESE| 0.23|\n", "| 1998| PETROBRAS| 276.0|\n", "+--------+------------+--------------+\n", "only showing top 20 rows\n" ] } ], "source": [ "from pyspark.sql.functions import col\n", "\n", "# select time and company attributes\n", "selTime = dimTime.select(\"keyTime\", \"yearTime\")\n", "selCompany = dimCompany.select(\"keyCompany\", \"nameCompany\")\n", "\n", "# join factStocks with time and company dimensions\n", "# - by default innerjoin\n", "# - only one copy of join column is kept\n", "joinedDF = (\n", " factStocks\n", " .join(selTime, factStocks[\"keyTime\"] == selTime[\"keyTime\"])\n", " .join(selCompany, \"keyCompany\")\n", " .select(\"yearTime\", \"nameCompany\", \"highValueStock\")\n", ")\n", "\n", "joinedDF.show()" ] }, { "cell_type": "code", "execution_count": 19, "id": "afe7a10f-f972-4318-bc28-2f41d6541ebf", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 38:==========================================================(2 + 0) / 2]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+--------+------------+------------------+---------+\n", "|yearTime| nameCompany| avg_stock|max_stock|\n", "+--------+------------+------------------+---------+\n", "| 2017| AZUL|26.732865226402712| 30.89|\n", "| 2019| COPHILLIPS| 61.20990576834049| 67.76|\n", "| 2019| VALERO ENER|161.76837264659792| 196.23|\n", "| 2019| ELECTR ARTS| 197.5250015258789| 215.12|\n", "| 2019| PETROCHIN|203.69400024414062| 210.3|\n", "| 2019| BAIDU INC| 504.7300109863281| 504.73|\n", "| 2020| HYPERA| 33.32437755783878| 42.8|\n", "| 2020| RANDON PART|10.637931720320001| 16.79|\n", "| 2020| VALE| 58.09566258809652| 88.58|\n", "| 2020| UNILEVER|302.50861499492936| 370.0|\n", "| 2020| AEGON NV| 18.73475000858307| 20.81|\n", "| 2016| SUL AMERICA| 17.21546188032771| 19.73|\n", "| 2016|BRISTOLMYERS| 218.8280554877387| 264.28|\n", "| 2017| ENERGIAS BR|14.316138201612768| 16.15|\n", "| 2017| BOSTON PROP| 39.52625012397766| 41.96|\n", "| 2018| LOG-IN| 4.248122458555261| 8.84|\n", "| 2019|CARREFOUR BR|21.097903267029793| 24.12|\n", "| 2019| US STEEL|54.037499809265135| 70.92|\n", "| 2019| BOOKING| 320.1116989423644| 339.22|\n", "| 2020| BNY MELLON|199.70409843569897| 226.0|\n", "+--------+------------+------------------+---------+\n", "only showing top 20 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql.functions import avg, max\n", "\n", "# agg is used when we want to apply multiple aggregate functions within a single operation\n", "joinedDF.groupBy(\"yearTime\", \"nameCompany\") \\\n", " .agg(\n", " avg(\"highValueStock\").alias(\"avg_stock\"),\n", " max(\"highValueStock\").alias(\"max_stock\")\n", " ) \\\n", " .show()" ] }, { "cell_type": "code", "execution_count": 20, "id": "a9c608cd-96ce-4449-97b4-7b9326b8a2f1", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 43:=============================> (1 + 1) / 2]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+--------+------------+------------------+---------+\n", "|yearTime| nameCompany| avg_stock|max_stock|\n", "+--------+------------+------------------+---------+\n", "| 2000| COELCE| 5.470495477453008| 7.5|\n", "| 2007| BRADESCO| 58.14718362457898| 90.26|\n", "| 2011| BRADESPAR|39.873456837218484| 48.99|\n", "| 2012| EQUATORIAL|15.482723581112497| 19.22|\n", "| 2014| LIGHT S/A|20.464233867583737| 25.49|\n", "| 2014|LOJAS RENNER| 68.30459663944859| 77.47|\n", "| 2015| COPHILLIPS| 48.10123361860003| 62.1|\n", "| 2015| GOL| 7.07260163334327| 15.33|\n", "| 2000| BAHEMA|38.142857142857146| 40.0|\n", "| 2000| EMBRAER|10.217217764546794| 13.35|\n", "| 2005| CEDRO| 74.42200063069662| 100.0|\n", "| 2009| USIMINAS|39.424369963203986| 56.7|\n", "| 2010| FERBASA|12.486441283039351| 18.0|\n", "| 2011| SANEPAR| 4.220326518039314| 4.98|\n", "| 2012| ELETROBRAS| 17.20189026022345| 27.59|\n", "| 2012| TECNISA| 8.77186991722603| 10.89|\n", "| 2013| PEPSICO INC| 185.1716651916504| 198.26|\n", "| 2015| MCDONALDS| 92.93075457159078| 118.88|\n", "| 2016| SUL AMERICA| 17.21546188032771| 19.73|\n", "| 2016|BRISTOLMYERS| 218.8280554877387| 264.28|\n", "+--------+------------+------------------+---------+\n", "only showing top 20 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "# registers the DataFrame as a temporary SQL view, which can then be queried using spark.sql()\n", "joinedDF.createOrReplaceTempView(\"joinedTable\")\n", "\n", "groupedDF = spark.sql(\"\"\"\n", " SELECT\n", " yearTime,\n", " nameCompany,\n", " avg(highValueStock) AS avg_stock,\n", " max(highValueStock) AS max_stock\n", " FROM joinedTable\n", " GROUP BY yearTime, nameCompany\n", "\"\"\")\n", "\n", "groupedDF.show()" ] }, { "cell_type": "code", "execution_count": 21, "id": "1de769aa-6959-4cc1-85af-3a4b1f90e56b", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 48:=============================> (1 + 1) / 2]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+--------+------------+--------------------+\n", "|yearTime| nameCompany| avg_quantity_stock|\n", "+--------+------------+--------------------+\n", "| 2000| COELCE| 688136.0855855856|\n", "| 2007| BRADESCO| 7.097130514081633E7|\n", "| 2011| BRADESPAR|1.8341864152263373E7|\n", "| 2012| EQUATORIAL| 5642397.711382113|\n", "| 2014| LIGHT S/A|1.6350280971774194E7|\n", "| 2014|LOJAS RENNER| 4.895360780241936E7|\n", "| 2015| COPHILLIPS| 18795.16883116883|\n", "| 2015| GOL|1.0293650158536585E7|\n", "| 2000| BAHEMA| 1420.0|\n", "| 2000| EMBRAER| 3211782.2983870967|\n", "| 2005| CEDRO| 55758.8|\n", "| 2009| USIMINAS| 7.41305579695122E7|\n", "| 2010| FERBASA| 1340546.537366548|\n", "| 2011| SANEPAR| 440752.1469387755|\n", "| 2012| ELETROBRAS|2.1358834022357725E7|\n", "| 2012| TECNISA| 6880398.434959349|\n", "| 2013| PEPSICO INC| 395361.75|\n", "| 2015| MCDONALDS| 462618.1509433962|\n", "| 2016| SUL AMERICA|1.3871265742971888E7|\n", "| 2016|BRISTOLMYERS| 311149.75|\n", "+--------+------------+--------------------+\n", "only showing top 20 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql.functions import avg\n", "\n", "# average stock quantity per year per company\n", "joinedQuantityDF = (\n", " factStocks\n", " .join(selTime, factStocks[\"keyTime\"] == selTime[\"keyTime\"])\n", " .join(selCompany, \"keyCompany\")\n", " .select(\"yearTime\", \"nameCompany\", \"quantityStock\")\n", " .groupBy(\"yearTime\", \"nameCompany\")\n", " .agg(avg(\"quantityStock\").alias(\"avg_quantity_stock\"))\n", ")\n", "\n", "joinedQuantityDF.show()\n", "\n", "joinedQuantityDF.createOrReplaceTempView(\"joinedQuantityTable\")" ] }, { "cell_type": "code", "execution_count": 22, "id": "59ae64d7-a022-4e7e-af9b-305948d5178a", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 53:=============================> (1 + 1) / 2]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-----------+--------+--------------------+\n", "|nameCompany|yearTime| avg_quantity_stock|\n", "+-----------+--------+--------------------+\n", "|RANDON PART| 2001| 35131.62162162162|\n", "|RANDON PART| 2002| 73362.47222222222|\n", "|RANDON PART| 2003| 145950.8458781362|\n", "|RANDON PART| 2004| 642942.7777777778|\n", "|RANDON PART| 2005| 1054024.5704225353|\n", "|RANDON PART| 2006| 2144967.1352040814|\n", "|RANDON PART| 2007| 3128629.170263789|\n", "|RANDON PART| 2008| 2574399.587804878|\n", "|RANDON PART| 2009| 2303429.385786802|\n", "|RANDON PART| 2010| 3883312.1052631577|\n", "|RANDON PART| 2011| 5052933.617021277|\n", "|RANDON PART| 2012| 7568332.294871795|\n", "|RANDON PART| 2013| 7826617.491271821|\n", "|RANDON PART| 2014| 5136473.757238307|\n", "|RANDON PART| 2015| 2168433.618644068|\n", "|RANDON PART| 2016| 3793190.205338809|\n", "|RANDON PART| 2017| 4771251.865957446|\n", "|RANDON PART| 2018| 8682911.583162218|\n", "|RANDON PART| 2019|1.1837507715151515E7|\n", "|RANDON PART| 2020|2.2096215779116467E7|\n", "+-----------+--------+--------------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql.functions import col\n", "\n", "resultDF = (\n", " joinedQuantityDF\n", " .filter(col(\"nameCompany\") == \"RANDON PART\")\n", " .select(\"nameCompany\", \"yearTime\", \"avg_quantity_stock\")\n", " .orderBy(col(\"yearTime\").asc())\n", ")\n", "\n", "resultDF.show()" ] }, { "cell_type": "code", "execution_count": 23, "id": "a2c13252-cb76-439d-bfca-b35d10da6a02", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 58:=============================> (1 + 1) / 2]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-----------+--------+--------------------+\n", "|nameCompany|yearTime| avg_quantity_stock|\n", "+-----------+--------+--------------------+\n", "|RANDON PART| 2001| 35131.62162162162|\n", "|RANDON PART| 2002| 73362.47222222222|\n", "|RANDON PART| 2003| 145950.8458781362|\n", "|RANDON PART| 2004| 642942.7777777778|\n", "|RANDON PART| 2005| 1054024.5704225353|\n", "|RANDON PART| 2006| 2144967.1352040814|\n", "|RANDON PART| 2007| 3128629.170263789|\n", "|RANDON PART| 2008| 2574399.587804878|\n", "|RANDON PART| 2009| 2303429.385786802|\n", "|RANDON PART| 2010| 3883312.1052631577|\n", "|RANDON PART| 2011| 5052933.617021277|\n", "|RANDON PART| 2012| 7568332.294871795|\n", "|RANDON PART| 2013| 7826617.491271821|\n", "|RANDON PART| 2014| 5136473.757238307|\n", "|RANDON PART| 2015| 2168433.618644068|\n", "|RANDON PART| 2016| 3793190.205338809|\n", "|RANDON PART| 2017| 4771251.865957446|\n", "|RANDON PART| 2018| 8682911.583162218|\n", "|RANDON PART| 2019|1.1837507715151515E7|\n", "|RANDON PART| 2020|2.2096215779116467E7|\n", "+-----------+--------+--------------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "resultDF = spark.sql(\"\"\"\n", " SELECT nameCompany, yearTime, avg_quantity_stock\n", " FROM joinedQuantityTable\n", " WHERE nameCompany = 'RANDON PART'\n", " ORDER BY yearTime ASC\n", "\"\"\")\n", "\n", "resultDF.show()" ] }, { "cell_type": "code", "execution_count": 24, "id": "bab7c3b5-c4a0-4706-9b0b-70dfcb65467a", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 63:=============================> (1 + 1) / 2]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------------+-----------------+---------+\n", "| nameCompany|number_of_records|max_stock|\n", "+------------+-----------------+---------+\n", "| JOHNSON| 1436| 891.8|\n", "| SANTANDER| 260| 18.52|\n", "| PLANOEPLANO| 71| 9.99|\n", "| TWITTER| 449| 152.4|\n", "| JPMORGAN| 1546| 302.65|\n", "| JSL| 2612| 36.81|\n", "|AMERICAN TOW| 60| 349.96|\n", "|BANK AMERICA| 1417| 153.56|\n", "| FRAS-LE| 1689| 20.0|\n", "| 3M| 1141| 242.69|\n", "| SCHULZ| 3796| 264.5|\n", "|CREDIT SUISS| 36| 34.89|\n", "| GRENDENE| 3998| 35.95|\n", "| UBS GROUP| 236| 81.0|\n", "| BP PLC| 39| 29.81|\n", "| ZOETIS INC| 49| 249.0|\n", "| BIONTECH SE| 43| 105.0|\n", "| VALID| 2509| 52.42|\n", "| ENERGISA| 2047| 60.44|\n", "|P.ACUCAR-CBD| 303| 130.0|\n", "+------------+-----------------+---------+\n", "only showing top 20 rows\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql.functions import count, max\n", "\n", "# count example\n", "joinedDF.groupBy(\"nameCompany\").agg(\n", " count(\"highValueStock\").alias(\"number_of_records\"),\n", " max(\"highValueStock\").alias(\"max_stock\")\n", ").show()" ] }, { "cell_type": "code", "execution_count": 25, "id": "3fc21bc9-ce3c-4282-a3f8-5bff0ee8039f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "== Physical Plan ==\n", "*(1) Filter (isnotnull(keyCompany#227) AND (keyCompany#227 = 1))\n", "+- FileScan csv [keyTime#226,keyCompany#227,openValueStock#228,closeValueStock#229,highValueStock#230,lowValueStock#231,quantityStock#232] Batched: false, DataFilters: [isnotnull(keyCompany#227), (keyCompany#227 = 1)], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/datasets/factStocks.csv], PartitionFilters: [], PushedFilters: [IsNotNull(keyCompany), EqualTo(keyCompany,1)], ReadSchema: struct\n", "\n", "\n" ] } ], "source": [ "# displaying physical plan\n", "factStocks.filter(col(\"keyCompany\") == 1).explain()\n", "factStocks.groupBy(\"keyCompany\").count().explain()" ] }, { "cell_type": "code", "execution_count": 26, "id": "11d6342e-e8b0-49e1-9dbe-89b694907388", "metadata": {}, "outputs": [], "source": [ "spark.stop()" ] } ], "metadata": { "kernelspec": { "display_name": "PySpark (Standalone Cluster)", "language": "python", "name": "pyspark" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.13" } }, "nbformat": 4, "nbformat_minor": 5 }