Libraries
","metadata":{"_kg_hide-output":true,"_kg_hide-input":true}},{"cell_type":"code","source":"import numpy as np \nimport pandas as pd \nimport os","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","_kg_hide-input":true,"execution":{"iopub.status.busy":"2022-06-02T18:06:35.651646Z","iopub.execute_input":"2022-06-02T18:06:35.652888Z","iopub.status.idle":"2022-06-02T18:06:35.68229Z","shell.execute_reply.started":"2022-06-02T18:06:35.652381Z","shell.execute_reply":"2022-06-02T18:06:35.681242Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"\n![](\"https://raw.githubusercontent.com/paulojunqueira/files/main/101.gif\")
\n\nThis is an introductory Pyspark hands on tutorial. I have decided to study and summarize some basic concepts of pyspark functions. Hope these can help some else!\n
","metadata":{}},{"cell_type":"markdown","source":"\n\n\n# CONTENT TABLE\n----------------\n## Concepts\n* [1) What is Spark \\ PySpark](#01)\n* [2) Characteristics](#03)\n \n\n## Practice\n* [1) GETTING RUNNING ON NOTEBOOKS ( Jupyter\\Colab\\Kaggle)](#1)\n* [2) READING DATA FROM DIFFERENT SOURCES](#2)\n* [3) DISPLAYING](#3)\n* [4) FILTERING](#4)\n* [5) FILTERING PART 2](#5)\n* [6) MANIPULATING DATA](#6)\n* [7) GROUPING BY](#7)\n* [8) WINDOW FUNCTION](#8)\n* [9) JOIN](#9)\n* [10) UDF](#10)\n* [11) REFERENCES](#20)\n\n## Next\n* 102 Pyspark - MLlib (Soon!) ","metadata":{}},{"cell_type":"markdown","source":"\n#What is Spark \\ PySpark
","metadata":{}},{"cell_type":"markdown","source":" \n Apache Spark is an open source unified computing engine for distributed data processing on computer clusters, allowing an easy and scalable developments. It supports many programing languages such as Python, R and Scala. Also, includes libraries from SQL and Machine Learning. The structure is designed to deal with huge amount of data distributing large datasets among \"computers\". Each one of the computers os called a \"Executors\" and the managemnt of all data is done by a so called \"Driver\".
\n
\n Pyspark is a library for python that enables to run python applications in the Apache Spark architeture, hence, allowing parallel distribuiton to cope with large data \n\n
![](\"https://raw.githubusercontent.com/paulojunqueira/files/main/Sem%20t%C3%ADtulo.gif\")
\n\n\n ","metadata":{}},{"cell_type":"markdown","source":"\n# Characteristics
\n\n\nAmong the characteristics of Spark, some interesting ones are:\n \n
GETTING RUNNING ON NOTEBOOKS ( Jupyter\\Colab\\Kaggle)
","metadata":{}},{"cell_type":"code","source":"!pip install pyspark -q","metadata":{"execution":{"iopub.status.busy":"2022-06-02T18:06:35.72114Z","iopub.execute_input":"2022-06-02T18:06:35.721956Z","iopub.status.idle":"2022-06-02T18:07:31.662266Z","shell.execute_reply.started":"2022-06-02T18:06:35.721916Z","shell.execute_reply":"2022-06-02T18:07:31.660847Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# To run pypsark in notebooks: \nfrom pyspark import SparkContext, SparkConf\nfrom pyspark.sql import SparkSession\n\n#Import libraries to create a dataframe\n#It is also common to import as T: #from pypsark.sql.types as T\nfrom pyspark.sql.types import StructType,StructField, StringType, IntegerType\n\n\nsc = SparkContext.getOrCreate(SparkConf().setMaster(\"local[*]\"))\nspark = SparkSession.builder.getOrCreate()","metadata":{"execution":{"iopub.status.busy":"2022-06-02T18:07:31.665707Z","iopub.execute_input":"2022-06-02T18:07:31.66602Z","iopub.status.idle":"2022-06-02T18:07:37.585764Z","shell.execute_reply.started":"2022-06-02T18:07:31.665981Z","shell.execute_reply":"2022-06-02T18:07:37.584073Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"\n#READING DATA FROM DIFFERENT SOURCES
\n\nDISPLAYING
\n\nFILTERING
\nFILTERING PART 2
","metadata":{}},{"cell_type":"code","source":"# Part 2\n\n#filtering Data\nprint('---- Example_6 Categorical ----')\ndf_t.filter((F.col('Embarked') == 'S') & (F.col('Sex') == 'female')).show(5)\n\n#filtering Data\nprint('---- Example_7 Exclusion ----')\ndf_t.filter(~(F.col('Embarked') == 'S')).show(5)\n\n#filtering Data\nprint('---- Example_8 Exclusion SQL ----')\ndf_t.filter(('Embarked != \"C\"')).show(5)\ndf_t.filter(('Embarked <> \"Q\"')).show(5)\n\n#filtering Data\nprint('---- Example_9 Cabin Starts with A----')\ndf_t.filter((F.col('Cabin').startswith('A'))).show(5)\n\n#filtering Data\nprint('---- Example_10 Cabin Ends with 3 ----')\ndf_t.filter((F.col('Cabin').endswith('3'))).show(5)\n\n#filtering Data\nprint('---- Example_11 Name Contains \"Miss\" ----')\ndf_t.filter((F.col('Name').contains('Miss'))).show(5)\n\n#filtering Data\nprint('---- Example_12 like (REGEX)----')\ndf_t.filter((F.col('Name').like('%John%'))).show(5)\n\n#filtering Data\nprint('---- Example_13 Cabin null----')\ndf_t.filter((F.col('Cabin').isNull())).show(5)\n\n#filtering Data\nprint('---- Example_14 Cabin not null----')\ndf_t.filter((F.col('Cabin').isNotNull())).show(5)","metadata":{"execution":{"iopub.status.busy":"2022-06-02T18:07:53.260554Z","iopub.execute_input":"2022-06-02T18:07:53.260928Z","iopub.status.idle":"2022-06-02T18:07:55.290635Z","shell.execute_reply.started":"2022-06-02T18:07:53.260881Z","shell.execute_reply":"2022-06-02T18:07:55.289481Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"\n#MANIPULATING DATA
\nGROUPING BY
","metadata":{}},{"cell_type":"code","source":"#Group By\nprint('---- Example_1 ---- Grouping by Sex and counting the total samples of each Genre')\ndf_t.groupBy('Sex').count().show()\n\n#Group By\nprint('---- Example_2 ---- Grouping by Survived and counting the total samples of each class')\ndf_t.groupBy('Survived').count().show()\n\n#Group By\nprint('---- Example_3 ---- Grouping by Sex and survived and counting the total samples of each class')\ndf_t.groupBy('Survived', 'Sex').count().show()\n\n#Group By\nprint('---- Example_3 ---- Grouping by Survived and get the age mean (avg)')\ndf_t.groupBy('Survived').agg(F.mean(F.col('Age'))).show()\n\n\n#Group By\nprint('---- Example_3.1 ---- Grouping by Survived and get the age mean (avg) and Std. Also(Changing the column´s name)')\ndf_t.groupBy('Survived').agg(F.mean(F.col('Age')).alias('Mean Age'),\n F.stddev(F.col('Age')).alias('Std Age')).show()\n\n#Group By\nprint('---- Example_4 ---- Grouping by quality')\ndf.groupBy('quality').agg(F.mean(F.col('fixed acidity')).alias('fixed acidity mean'),\n F.stddev(F.col('fixed acidity')).alias('Std fixed acidity'),\n F.mean(F.col('residual sugar')).alias('residual sugar mean'),\n F.stddev(F.col('residual sugar')).alias('Std residual sugar')).show()\n\n#Group By\nprint('---- Example_5 ---- Grouping age and counting only age > 20')\ndf_t.groupBy('Age').agg(F.when(F.col('Age') > 20, \\\n F.count(F.col('Age'))).alias('Count Age > 20')).show()","metadata":{"execution":{"iopub.status.busy":"2022-06-02T18:07:56.860315Z","iopub.execute_input":"2022-06-02T18:07:56.860649Z","iopub.status.idle":"2022-06-02T18:07:59.352424Z","shell.execute_reply.started":"2022-06-02T18:07:56.860604Z","shell.execute_reply":"2022-06-02T18:07:59.351481Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"\n#WINDOW FUNCTION
","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.window import Window\n\n#Creating a sequencial column in the time window\nprint('---- Example_1 ---- Creating a sequential Row')\nwindow = Window.partitionBy(\"quality\").orderBy(\"quality\")\n\ndf_temp = df.withColumn(\"ROW COLUMN\",F.row_number().over(window))\ndf_temp.select('quality','alcohol','ROW COLUMN').show()\n\n#suming alcohol values over the quality variable\nprint('---- Example_2 ---- Sum over sequence')\nwindow = Window.partitionBy(\"quality\").orderBy(\"quality\")\n\ndf_temp = df.withColumn(\"sum\",F.sum(F.col('alcohol')).over(window))\ndf_temp.select('quality','alcohol','sum').show()\n\n\n#Cumulative sum over alcohol by quality\nprint('---- Example_3 ---- Cumulative Sum over alcohol with ROUND')\nwindow = Window.partitionBy(\"quality\").orderBy(\"alcohol\").rangeBetween(Window.unboundedPreceding, 0)\n\ndf_temp = df.withColumn(\"CUMULATIVE_alcohol\",F.round(F.sum(F.col('alcohol')).over(window),1))\ndf_temp.select('quality','alcohol','CUMULATIVE_alcohol').show()\n\n#Laging\nprint('---- Example_5 ---- Laging alcohol varibale by 2 (moving down 2 positions)')\nwindow = Window.partitionBy(\"quality\").orderBy(\"alcohol\")\n\ndf_temp = df.withColumn(\"Lagged_alcohol\",F.lag(F.col('alcohol'),2).over(window))\ndf_temp.select('quality','alcohol','Lagged_alcohol').show()\n\n#Lead\nprint('---- Example_6 ---- Lead alcohol varibale by 2 (moving up 2 positions)')\nwindow = Window.partitionBy(\"quality\").orderBy(\"alcohol\")\n\ndf_temp = df.withColumn(\"leaded_alcohol\",F.lead(F.col('alcohol'),2).over(window))\ndf_temp.select('quality','alcohol','leaded_alcohol').show()","metadata":{"execution":{"iopub.status.busy":"2022-06-02T18:07:59.353638Z","iopub.execute_input":"2022-06-02T18:07:59.354814Z","iopub.status.idle":"2022-06-02T18:08:01.322643Z","shell.execute_reply.started":"2022-06-02T18:07:59.354765Z","shell.execute_reply":"2022-06-02T18:08:01.321678Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"\n#JOIN
\n\nUDF
\n\nUser Defined Functions or UDF are a SQL wrapper of python functions. Hence, it is possible to create python functions and apply to transformer columns of a pyspark df
","metadata":{}},{"cell_type":"code","source":"\n# Let's say that we want to remove the F3 of column C5 of df_union previously created\n# First, create a dummy function. \n\ndef remove_F3(x):\n \"\"\"Function to split string and return first element\"\"\"\n return x.split('_')[0]\n\nprint(f'Teste String ABC_F3 -> {remove_F3(\"ABC_F3\")}\\n')\n\n# Next wrapping the functions into a udf(). The second arg is the type of the return element\nremove_F3_udf = F.udf(lambda x: remove_F3(x),StringType())\n\n#Finally, transform the C5 column\nprint('--- Example_1 --- Removing the F3 string from C5 Column with a custom function')\ndf_union_t = df_union.withColumn('C5_tranformed', remove_F3_udf(F.col('C5')))\ndf_union_t.show()","metadata":{"execution":{"iopub.status.busy":"2022-06-02T19:37:24.628313Z","iopub.execute_input":"2022-06-02T19:37:24.62863Z","iopub.status.idle":"2022-06-02T19:37:24.986219Z","shell.execute_reply.started":"2022-06-02T19:37:24.628593Z","shell.execute_reply":"2022-06-02T19:37:24.985562Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\n# Let's say that we want to combine the newly C5_tranformed (string) with a C6 (Number) multiplied by a factor (Why not?)\n# First, create a dummy function. \n\ndef combination_Func(st, num, factor):\n \"\"\"Combine a string (st) with a num times a factor\"\"\"\n return st+'_'+str(round(num*factor))\n\nprint(f'Teste String ABC \\ Number 10 \\ Factor 0.2 -> {combination_Func(\"ABC_F3\",10,0.2)}')\n\n# Next wrapping the functions into a udf(). The second arg is the type of the return element\ncombination_Func_udf = F.udf(lambda x,y,z: combination_Func(x,y,z),StringType())\n\n#Finally, transform the C5 column\nprint('--- Example_2 --- Creating a new column C7 based on C5_tranformed and C6 and function combination_Func')\ndf_union_t.withColumn('C7', combination_Func_udf(F.col('C5_tranformed'),F.col('C6'), F.lit(0.2))).show()","metadata":{"execution":{"iopub.status.busy":"2022-06-02T19:54:34.692249Z","iopub.execute_input":"2022-06-02T19:54:34.692566Z","iopub.status.idle":"2022-06-02T19:54:35.238815Z","shell.execute_reply.started":"2022-06-02T19:54:34.692534Z","shell.execute_reply":"2022-06-02T19:54:35.237878Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"\n#REFERENCES
\n\n* [Spark Apache](https://spark.apache.org/)\n* [Spark Documentation](https://spark.apache.org/docs/2.4.0/api/python/)\n* [Spark SQL Functions](https://spark.apache.org/docs/2.4.0/api/python/pyspark.sql.html#pyspark.sql.functions)\n* (Book) Spark: The Definitive Guide: Big Data Processing Made Simple\n* (Book) Essential PySpark for Scalable Data Analytics: A beginner's guide to harnessing the power and ease of PySpark 3","metadata":{}},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}