PySpark_Intro.ipynb

{"cells":[{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"cc4b8a83-3b0c-4b0f-ad46-194294890ad2","showTitle":false,"title":""},"id":"ICXbyhtyjXZM"},"source":["#**Introduction to PySpark**\n","This is an introduction to Spark DataFrames and MLlib in Python.\n","\n","\n"]},{"cell_type":"code","source":["# Clone the GitHub repository\n","!git clone https://github.com/ssalloum/SDSC-Spark4.git"],"metadata":{"id":"-LlkdCTePkOd","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121736272,"user_tz":-480,"elapsed":1965,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"4e915a4f-184d-4685-9487-defb89077554"},"execution_count":16,"outputs":[{"output_type":"stream","name":"stdout","text":["Cloning into 'SDSC-Spark4'...\n","remote: Enumerating objects: 28, done.\u001b[K\n","remote: Counting objects: 100% (28/28), done.\u001b[K\n","remote: Compressing objects: 100% (15/15), done.\u001b[K\n","remote: Total 28 (delta 4), reused 22 (delta 4), pack-reused 0\u001b[K\n","Receiving objects: 100% (28/28), 7.24 MiB | 12.27 MiB/s, done.\n","Resolving deltas: 100% (4/4), done.\n"]}]},{"cell_type":"code","source":["!ls /content/SDSC-Spark4/"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"dDt-vQnbPqi1","executionInfo":{"status":"ok","timestamp":1692121748051,"user_tz":-480,"elapsed":380,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"325136fc-57e0-4ebd-92a6-303e241855af"},"execution_count":18,"outputs":[{"output_type":"stream","name":"stdout","text":["data  README.md\n"]}]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"120286d4-6e58-4a7a-8294-fe6bbb46e215","showTitle":false,"title":""},"id":"cDefeMsvjXZR"},"source":["# PySpark\n","PySpark is an interface for Apache Spark that allows users to write Spark applications using python APIs. PySpark supports most of Spark’s features such as Spark SQL, Streaming, MLlib (Machine Learning) and Spark Core. For detailed information on these components and APIs, please refer to the [official PySpark Documentation](https://spark.apache.org/docs/latest/api/python/index.html)."]},{"cell_type":"code","source":["#You don't need this on Databricks\n","!pip install pyspark"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"u04pguFNlZvy","executionInfo":{"status":"ok","timestamp":1692121754858,"user_tz":-480,"elapsed":6375,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"00abc8c1-9b1f-42cc-c5f6-bf1528a5858d"},"execution_count":19,"outputs":[{"output_type":"stream","name":"stdout","text":["Requirement already satisfied: pyspark in /usr/local/lib/python3.10/dist-packages (3.4.1)\n","Requirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from pyspark) (0.10.9.7)\n"]}]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"41df5dad-1f3e-4c84-ad6d-c2700ad4c3f2","showTitle":false,"title":""},"id":"2BTx92K_jXZR"},"source":["## Spark SQL\n","* [Spark SQL](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/index.html) is a Spark module for structured data processing.\n","* Spark SQL integrates relational processing (using SQL) and functional programming (using the DataFrame API).\n","\n"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"e0f40f1b-2c24-4e08-90ad-a4924e159ec4","showTitle":false,"title":""},"id":"FWxdNXxrjXZS"},"source":["## SparkSession\n","* An essentail class in Spark SQL is [pyspark.sql.SparkSession](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.SparkSession.html) which represents a unified entry point to programming in Spark.\n","\n","* In Spark-shell or Databricks notebooks, a SparkSession is created for you, stored in a variable called `spark`."]},{"cell_type":"code","source":["# You don't need this on Databricks or spark-shell\n","from pyspark.sql import SparkSession\n","\n","# Create a Spark Session\n","spark = SparkSession.builder\\\n","        .master(\"local[*]\")\\\n","        .appName(\"Intro to PySpark\")\\\n","        .getOrCreate()"],"metadata":{"id":"1WwCBTK_ljcw","executionInfo":{"status":"ok","timestamp":1692121754860,"user_tz":-480,"elapsed":37,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":20,"outputs":[]},{"cell_type":"code","execution_count":21,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"4b5a5bd0-fba9-4d06-84ef-fb36c9e6e1ca","showTitle":false,"title":""},"id":"yxZdi3iDjXZS","colab":{"base_uri":"https://localhost:8080/","height":219},"executionInfo":{"status":"ok","timestamp":1692121754860,"user_tz":-480,"elapsed":34,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"af36f674-4a59-4aac-fd57-10e1cb530a72"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["<pyspark.sql.session.SparkSession at 0x7acc61fecfa0>"],"text/html":["\n","            <div>\n","                <p><b>SparkSession - in-memory</b></p>\n","                \n","        <div>\n","            <p><b>SparkContext</b></p>\n","\n","            <p><a href=\"http://7010f70f5adb:4040\">Spark UI</a></p>\n","\n","            <dl>\n","              <dt>Version</dt>\n","                <dd><code>v3.4.1</code></dd>\n","              <dt>Master</dt>\n","                <dd><code>local[*]</code></dd>\n","              <dt>AppName</dt>\n","                <dd><code>Intro to PySpark</code></dd>\n","            </dl>\n","        </div>\n","        \n","            </div>\n","        "]},"metadata":{},"execution_count":21}],"source":["# Check Spark Session Information\n","spark"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"badc07b7-9836-4f88-b084-728f24af9567","showTitle":false,"title":""},"id":"RGS2gYttjXZU"},"source":["## SparkContext\n","* [SparkContext](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkContext.html) was the main entry point in earlier versions of Spark.\n","* For working with low-level APIs, [Resilient Distributed Datasets (RDDs)](https://spark.apache.org/docs/latest/rdd-programming-guide.html), and for backward compatibility, you can access SparkContext via SparkSession."]},{"cell_type":"code","execution_count":22,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"292a260e-336d-4e4a-949b-e59d16e6d014","showTitle":false,"title":""},"id":"dQmpAI-ujXZV","colab":{"base_uri":"https://localhost:8080/","height":196},"executionInfo":{"status":"ok","timestamp":1692121754861,"user_tz":-480,"elapsed":32,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"ae0e46c8-2ee7-4eab-d12e-b83e4f408703"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["<SparkContext master=local[*] appName=Intro to PySpark>"],"text/html":["\n","        <div>\n","            <p><b>SparkContext</b></p>\n","\n","            <p><a href=\"http://7010f70f5adb:4040\">Spark UI</a></p>\n","\n","            <dl>\n","              <dt>Version</dt>\n","                <dd><code>v3.4.1</code></dd>\n","              <dt>Master</dt>\n","                <dd><code>local[*]</code></dd>\n","              <dt>AppName</dt>\n","                <dd><code>Intro to PySpark</code></dd>\n","            </dl>\n","        </div>\n","        "]},"metadata":{},"execution_count":22}],"source":["# get SparkContext\n","sc = spark.sparkContext\n","sc"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"e5a9517b-fa46-428c-9c36-41761abe891b","showTitle":false,"title":""},"id":"1DIcPKJLjXZW"},"source":["# Spark DataFrames\n","\n","\n","*   [pyspark.sql.DataFrame](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.html) represents a distributed collection of data grouped into named columns.\n","\n","* [pyspark.sql.Column](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.Column.html): represents a column in a DataFrame.\n","* [pyspark.sql.Row](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.Row.html): represents a row in a DataFrame.\n","*   [pyspark.sql.functions](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/functions.html): common functions to work with DataFrames.\n","\n","* A DataFrame can be constructed from a variety of [supported data sources](https://spark.apache.org/docs/latest/sql-data-sources.html).\n"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{},"inputWidgets":{},"nuid":"dd7427b7-0be6-4411-8f00-7461915a5184","showTitle":false,"title":""},"id":"x6PQ_t_uaiof"},"source":["## DataFrameReader\n","* [DataFrameReader](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.html): loading DataFrames from external sources.\n","* You cannot create an instance of DataFrameReader.\n","* You can access a DataFrameReader through a SparkSession instance using the [read](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.SparkSession.read.html) property to read data from a static data source (streaming data sources has a different method: readStream).\n","* DataFrameReader provides several public methods that can be used with all supported data sources, and may take different arguments for each source: [format](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.format.html), [option](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.option.html), [schema](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.schema.html), and [load](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.load.html).\n","\n","* If you don’t specify the format, then the default is\n","Parquet or whatever is set in 'spark.sql.sources.default'.\n","\n","* DataFrameReader also has methods to directly load data from specific formats/sources such as [parquet](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.parquet.html), [csv](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.csv.html), [json](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.json.html)."]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"490f371f-cd83-4d86-9541-2ddd7006da55","showTitle":false,"title":""},"id":"QkBIwNB7SvaM"},"source":["### Creating DataFrames From CSV files\n","* You can read data from a [CSV file](https://spark.apache.org/docs/latest/sql-data-sources-csv.html) into a DataFrame.\n","* The [pyspark.sql.SparkSession.read](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.SparkSession.read.html) function can be used to read in the CSV file and returns a DataFrame of rows and named columns with the types dictated in the schema. We will use csv files from the flights dataset:"]},{"cell_type":"code","source":["dataPath = \"/content/SDSC-Spark4/data/2015-summary.csv\""],"metadata":{"id":"80rCxGpsWFGO","executionInfo":{"status":"ok","timestamp":1692121754861,"user_tz":-480,"elapsed":31,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":23,"outputs":[]},{"cell_type":"code","execution_count":24,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"4897315a-ef2a-468b-b0f3-972e1a6174b1","showTitle":false,"title":""},"id":"M9GYMrValVUR","executionInfo":{"status":"ok","timestamp":1692121756786,"user_tz":-480,"elapsed":1955,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"outputs":[],"source":["#try with inferSchema\n","flights_df = spark.read\\\n","  .option(\"inferSchema\", \"true\")\\\n","  .option(\"header\", \"true\")\\\n","  .csv(dataPath)"]},{"cell_type":"code","execution_count":25,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"157538b9-0046-477f-8670-f8109fcb93dd","showTitle":false,"title":""},"id":"nnYClcJujXZX","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121757170,"user_tz":-480,"elapsed":394,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"8de5f4ef-7509-4c66-ef2e-a4f03365d7ed"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-------------------+-----+\n","|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n","+-----------------+-------------------+-----+\n","|    United States|            Romania|   15|\n","|    United States|            Croatia|    1|\n","|    United States|            Ireland|  344|\n","|            Egypt|      United States|   15|\n","|    United States|              India|   62|\n","+-----------------+-------------------+-----+\n","only showing top 5 rows\n","\n"]}],"source":["flights_df.show(5)"]},{"cell_type":"code","execution_count":26,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"ff2d0858-c984-4497-9e78-dcf853cc9b49","showTitle":false,"title":""},"id":"-GnXDJGrjXZX","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121758076,"user_tz":-480,"elapsed":911,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"f3be258a-3666-4201-9d9a-ea0d36d06b60"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["256"]},"metadata":{},"execution_count":26}],"source":["flights_df.count()"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"0b7a4743-1bd2-4f22-a3aa-708cffeafd2a","showTitle":false,"title":""},"id":"UYdWGEmHC0q3"},"source":["## DataFrame Schema\n","* A schema in Spark defines the column names and associated data types for a DataFrame. In addition to inferring the schema from the source data, Spark allows you to define a schema programmatically.\n","\n","* A schema is a [StructType](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.types.StructType.html) made up of a number of fields, each field is a [StructField](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.types.StructField.htm), that have a name, type, a Boolean flag which specifies whether that column can contain missing or null values, and, finally, users can optionally specify associated metadata with that column.\n","*  Supported data types are defined in [pyspark.sql.types](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/data_types.html).\n","\n","* [Spark SQL Guide: Data Types](https://spark.apache.org/docs/latest/sql-ref-datatypes.html)"]},{"cell_type":"code","execution_count":27,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"2f8048b1-7a7e-433d-a54c-76a5167cf32b","showTitle":false,"title":""},"id":"Pm-iq_EgjXZY","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121758077,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"f2370f67-0dc0-4bb4-a8fc-cf4aef8046ab"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["StructType([StructField('dest', StringType(), True), StructField('origin', StringType(), True), StructField('flights', LongType(), False)])"]},"metadata":{},"execution_count":27}],"source":["#Define a schema programatically\n","from pyspark.sql.types import *\n","\n","myFlightSchema = StructType([\n","  StructField(\"dest\", StringType(), True),\n","  StructField(\"origin\", StringType(), True),\n","  StructField(\"flights\", LongType(), False)\n","])\n","\n","myFlightSchema"]},{"cell_type":"code","execution_count":28,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"54f12595-a93b-42db-b379-fc022d6d9849","showTitle":false,"title":""},"id":"j97dhRk_YzZ3","executionInfo":{"status":"ok","timestamp":1692121758077,"user_tz":-480,"elapsed":6,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"outputs":[],"source":["flights_df_2015 = spark.read.schema(myFlightSchema).option(\"header\", \"true\").csv(dataPath)"]},{"cell_type":"code","source":["flights_df_2015.take(5)"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"dYT75KqnPg85","executionInfo":{"status":"ok","timestamp":1692121758462,"user_tz":-480,"elapsed":391,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"e6d6a40e-6ebc-4fc5-8b55-731389f5256f"},"execution_count":29,"outputs":[{"output_type":"execute_result","data":{"text/plain":["[Row(dest='United States', origin='Romania', flights=15),\n"," Row(dest='United States', origin='Croatia', flights=1),\n"," Row(dest='United States', origin='Ireland', flights=344),\n"," Row(dest='Egypt', origin='United States', flights=15),\n"," Row(dest='United States', origin='India', flights=62)]"]},"metadata":{},"execution_count":29}]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"67a5f652-e418-4755-a45b-1f3b571c8e97","showTitle":false,"title":""},"id":"H32nISfLvzLV"},"source":["## Columns\n","* [DataFrame.columns](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.columns.html): get all columns names in a DataFrame.\n","\n"]},{"cell_type":"code","execution_count":30,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"e6f3a6ed-9cd6-405e-8f8f-784e697775a7","showTitle":false,"title":""},"id":"nl9_0lIacbow","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121758832,"user_tz":-480,"elapsed":378,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"1d14a9a6-7f3b-447d-9af6-8b8cd76d55b7"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["['dest', 'origin', 'flights']"]},"metadata":{},"execution_count":30}],"source":["flights_df_2015.columns"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"3dc7e416-b0fc-469f-91b6-1fbc696c31d5","showTitle":false,"title":""},"id":"nWWZGAuxnuyI"},"source":["* you can refer to columns in a number of different\n","ways; and you can use them interchangeably: [col()](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.col.html), [column()](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.column.html), [expr()](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.expr.html)."]},{"cell_type":"code","execution_count":31,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"b547d08e-aab6-49c2-a3bf-ed4d2f2dca68","showTitle":false,"title":""},"id":"fgt7SLbxnzkR","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121759230,"user_tz":-480,"elapsed":408,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"2c2589d3-24d2-40ce-c9d6-32a371f5b7cd"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------+-------------+-------------+-------------+\n","|         dest|         dest|  lower(dest)|         dest|\n","+-------------+-------------+-------------+-------------+\n","|United States|United States|united states|United States|\n","|United States|United States|united states|United States|\n","|United States|United States|united states|United States|\n","|        Egypt|        Egypt|        egypt|        Egypt|\n","|United States|United States|united states|United States|\n","+-------------+-------------+-------------+-------------+\n","only showing top 5 rows\n","\n"]}],"source":["from pyspark.sql.functions import expr, col, column\n","\n","flights_df_2015.select(\n","  col(\"dest\"),\n","  column(\"dest\"),\n","  expr(\"lower(dest)\"),\n","  flights_df_2015.dest)\\\n",".show(5)"]},{"cell_type":"markdown","source":["* In Spark DataFrames, Columns are objects represented by [pyspark.sql.Column](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.Column.html) that provides commonly used methods on columns."],"metadata":{"id":"-pKfoXkY1CuM"}},{"cell_type":"code","source":["from pyspark.sql import Column\n","\n","flights_df_2015.orderBy(flights_df_2015.flights.desc()).show(5)"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"1CBrSp4q1M4U","executionInfo":{"status":"ok","timestamp":1692121759966,"user_tz":-480,"elapsed":756,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"b3951e35-2f16-497a-e6fd-3a905441b295"},"execution_count":32,"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------+-------------+-------+\n","|         dest|       origin|flights|\n","+-------------+-------------+-------+\n","|United States|United States| 370002|\n","|United States|       Canada|   8483|\n","|       Canada|United States|   8399|\n","|United States|       Mexico|   7187|\n","|       Mexico|United States|   7140|\n","+-------------+-------------+-------+\n","only showing top 5 rows\n","\n"]}]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"26a88f43-a58b-44d6-824b-46083f5d9dc3","showTitle":false,"title":""},"id":"v9YMW9_Lv2GY"},"source":["## Rows\n","* A row in Spark is an object of [pyspark.sql.Row](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.Row.html), containing one or more columns.\n"]},{"cell_type":"code","source":["#get the first Row\n","flights_df_2015.first()"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"1h5CTsbM72Ap","executionInfo":{"status":"ok","timestamp":1692121760367,"user_tz":-480,"elapsed":404,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"cfa2d354-8549-4784-ed15-0bad47489209"},"execution_count":33,"outputs":[{"output_type":"execute_result","data":{"text/plain":["Row(dest='United States', origin='Romania', flights=15)"]},"metadata":{},"execution_count":33}]},{"cell_type":"code","source":["#get a list of the first \"num\" of Rows"],"metadata":{"id":"CG8d38oL-j08","executionInfo":{"status":"ok","timestamp":1692121760369,"user_tz":-480,"elapsed":15,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":34,"outputs":[]},{"cell_type":"code","source":["flights_df_2015.take(5)"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"FFfX8G30-sUx","executionInfo":{"status":"ok","timestamp":1692121760371,"user_tz":-480,"elapsed":16,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"6a5bc685-a1a3-4a76-91d6-ee60bd9d12c6"},"execution_count":35,"outputs":[{"output_type":"execute_result","data":{"text/plain":["[Row(dest='United States', origin='Romania', flights=15),\n"," Row(dest='United States', origin='Croatia', flights=1),\n"," Row(dest='United States', origin='Ireland', flights=344),\n"," Row(dest='Egypt', origin='United States', flights=15),\n"," Row(dest='United States', origin='India', flights=62)]"]},"metadata":{},"execution_count":35}]},{"cell_type":"markdown","source":["* Because Row is an object in Spark and an ordered collection of fields, you can instantiate a Row in each of Spark’s supported languages and access its fields by an index starting at 0:"],"metadata":{"id":"RqFnrMZp-fUV"}},{"cell_type":"code","execution_count":36,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"7bf4d47b-d823-424f-86a3-496ad4a82275","showTitle":false,"title":""},"id":"n-uB1JHxOt2p","colab":{"base_uri":"https://localhost:8080/","height":35},"executionInfo":{"status":"ok","timestamp":1692121760375,"user_tz":-480,"elapsed":16,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"4f5a29f5-788c-42a7-b410-0bd848b16a4e"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["'Reynold'"],"application/vnd.google.colaboratory.intrinsic+json":{"type":"string"}},"metadata":{},"execution_count":36}],"source":["from pyspark.sql import Row\n","\n","blog_row = Row(6, \"Reynold\", \"Xin\", \"https://tinyurl.6\", 255568, \"3/2/2015\",\n","[\"twitter\", \"LinkedIn\"])\n","\n","# access using index for individual items\n","blog_row[1]\n","'Reynold'"]},{"cell_type":"code","execution_count":37,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"4edfb72c-3cde-4e2f-afc2-666159714783","showTitle":false,"title":""},"id":"rPAkXJ-fFj3m","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121760735,"user_tz":-480,"elapsed":375,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"b1b8df6d-fbeb-4a36-e505-fda099565af9"},"outputs":[{"output_type":"stream","name":"stdout","text":["+---+\n","| id|\n","+---+\n","|  0|\n","|  1|\n","|  2|\n","|  3|\n","|  4|\n","+---+\n","\n"]}],"source":["# the following code results in an array of Row objects.\n","spark.range(5).show()"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"d7898a94-3d52-4c28-b771-80f3b15938bd","showTitle":false,"title":""},"id":"3Dn93q7WPaz8"},"source":["Row objects can be used to create DataFrames if you need them for quick interactivity\n","and exploration:"]},{"cell_type":"code","execution_count":38,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"7c6f8e91-9341-4d39-be4a-e62eb3aa5024","showTitle":false,"title":""},"id":"bntpp8YuPcSU","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121767714,"user_tz":-480,"elapsed":6981,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"e15230c9-ed00-480f-941d-1856e0e39c3e"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------+-----+\n","|      Authors|State|\n","+-------------+-----+\n","|Matei Zaharia|   CA|\n","|  Reynold Xin|   CA|\n","+-------------+-----+\n","\n","root\n"," |-- Authors: string (nullable = true)\n"," |-- State: string (nullable = true)\n","\n"]}],"source":["rows = [Row(\"Matei Zaharia\", \"CA\"), Row(\"Reynold Xin\", \"CA\")]\n","authors_df = spark.createDataFrame(rows, [\"Authors\", \"State\"])\n","authors_df.show()\n","authors_df.printSchema()"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{},"inputWidgets":{},"nuid":"bfb593c2-e1be-47e9-821f-518ca9720a2c","showTitle":false,"title":""},"id":"7RbY0UI6YPLv"},"source":["## Parquet Data Source\n","* [Parquet](https://parquet.apache.org/) is an open-source columnar format that offers many I/O\n","optimizations (such as compression, which saves storage space and allows for quick\n","access to data columns).\n","\n","* [Parquet files](https://github.com/apache/parquet-format#file-format) are stored in a directory structure that contains the data files, metadata,\n","a number of compressed files, and some status files.\n","\n","* Spark SQL provides support for [reading and writing Parquet files](https://spark.apache.org/docs/latest/sql-data-sources-parquet.html).\n","\n","* Parquet is the default data\n","source in Spark.\n","\n","* Unless you are reading from a streaming data source, there’s no need to supply a\n","schema when reading from a Parquet file, because Parquet saves it as part of its metadata.\n","\n","* Another way to read this same data using the [parquet](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.parquet.html) method."]},{"cell_type":"code","source":["parquetPath = \"/content/SDSC-Spark4/data/2010-summary.parquet\""],"metadata":{"id":"jWsUAIBWldTZ","executionInfo":{"status":"ok","timestamp":1692121807579,"user_tz":-480,"elapsed":345,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":41,"outputs":[]},{"cell_type":"code","execution_count":42,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"3e042e91-c6cc-446a-a8b5-cdd5cfd6e65d","showTitle":false,"title":""},"id":"fRNL7MdUYPLw","executionInfo":{"status":"ok","timestamp":1692121809132,"user_tz":-480,"elapsed":4,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"outputs":[],"source":["df2 = spark.read.parquet(parquetPath)"]},{"cell_type":"code","source":["df2.printSchema()"],"metadata":{"id":"dQsJXsG3nTPS","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121812922,"user_tz":-480,"elapsed":342,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"57621992-b76a-4663-c8cc-db3eaccb1b0a"},"execution_count":43,"outputs":[{"output_type":"stream","name":"stdout","text":["root\n"," |-- DEST_COUNTRY_NAME: string (nullable = true)\n"," |-- ORIGIN_COUNTRY_NAME: string (nullable = true)\n"," |-- count: long (nullable = true)\n","\n"]}]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{},"inputWidgets":{},"nuid":"8484a997-ee02-44db-b583-fbbef4d54f1f","showTitle":false,"title":""},"id":"WmcTB5XralNE"},"source":["## DataFrameWriter\n","* [DataFrameWriter](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.html) is an interface used to write a DataFrame to external stoage systems.\n","\n","* Unlike with DataFrameReader, you access its instance not from a SparkSession but from the DataFrame you wish to save.\n","\n","* To get an instance handle, use the [DataFrame.write](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.write.html) method for static data sources (DataFrame.writeStream for streaming data sources).\n","\n","* It also provides several public methods: [format](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.format.html), [option](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.option.html), [bucketBy](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.bucketBy.html), [save](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.save.html), and [saveAsTable](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.saveAsTable.html).\n","* DataFrameWriter also has methods to directly write data to specific formats/sources such as [parquet](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.parquet.html), [csv](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.csv.html), [json](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.json.html)."]},{"cell_type":"code","execution_count":44,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{},"inputWidgets":{},"nuid":"d5f3604f-d2c1-4b27-a804-cb15810f4e28","showTitle":false,"title":""},"id":"pqZeB5diLn0z","executionInfo":{"status":"ok","timestamp":1692121840689,"user_tz":-480,"elapsed":1627,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"outputs":[],"source":["df2.write.parquet(path=\"/tmp/data/df_parquet1\",\n","  mode=\"overwrite\",\n","  compression=\"snappy\")"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"5eef23ee-4983-48a0-9211-86edd2340eb0","showTitle":false,"title":""},"id":"l-6HL_siq6XE"},"source":["#DataFrame Operations: Transformations and Actions\n","\n","* Spark operations on DataFrames can be classified into two types: transformations and actions.\n","* All transformations are evaluated lazily. Their results are not computed immediately,\n","but they are recorded as a lineage. This allows Spark to optimize the execution\n","plan.\n","* Distributed computation occurs upon invoking an action on a DataFrame, e.g.,: `show(), take(), count(), collect()`."]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"9666ef6a-1d65-4b60-89c2-6aa1126d2b12","showTitle":false,"title":""},"id":"NDkcHGzHln1_"},"source":["### select\n","The easiest way to work with columns is just to use the [select](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.select.html) method and pass in the column names as strings:"]},{"cell_type":"code","execution_count":46,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"ec55a4dd-4da1-4bb0-8124-0eb0e754a08a","showTitle":false,"title":""},"id":"p4-uudqsmmDy","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121886278,"user_tz":-480,"elapsed":326,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"0c1a7584-5392-4545-d7d5-7d3981f42b5e"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+\n","|DEST_COUNTRY_NAME|\n","+-----------------+\n","|    United States|\n","|    United States|\n","+-----------------+\n","only showing top 2 rows\n","\n"]}],"source":["flights_df.select(\"DEST_COUNTRY_NAME\").show(2)"]},{"cell_type":"code","execution_count":47,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"63677a45-744b-4fc7-8291-9290df25f0bc","showTitle":false,"title":""},"id":"JjgfTWv9m-o4","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121887545,"user_tz":-480,"elapsed":490,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"1da70e12-110a-43ec-b7ff-ec9bb6d27006"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-------------------+\n","|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|\n","+-----------------+-------------------+\n","|    United States|            Romania|\n","|    United States|            Croatia|\n","+-----------------+-------------------+\n","only showing top 2 rows\n","\n"]}],"source":["flights_df.select(\"DEST_COUNTRY_NAME\",\"ORIGIN_COUNTRY_NAME\").show(2)"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"5dac23b4-80fd-4c12-ad6d-e376ca137c82","showTitle":false,"title":""},"id":"LWgIpyw2lp6n"},"source":["### selectExpr\n","Because select followed by a series of expr is such a common pattern, Spark has a shorthand for doing this efficiently: [selectExpr](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.selectExpr.html)."]},{"cell_type":"code","execution_count":48,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"67f91f27-8a70-430f-9845-fb85cba26b5f","showTitle":false,"title":""},"id":"ev0edy3zndoW","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121887886,"user_tz":-480,"elapsed":348,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"f70ca394-6e88-4255-d296-60cb3e752ea2"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------+-------+\n","|  Destination| Origin|\n","+-------------+-------+\n","|United States|Romania|\n","|United States|Croatia|\n","+-------------+-------+\n","only showing top 2 rows\n","\n"]}],"source":["flights_df.selectExpr(\"DEST_COUNTRY_NAME as Destination\", \"Origin_COUNTRY_NAME as Origin\").show(2)"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"35339b1d-7e11-4663-b546-cd009dab86fd","showTitle":false,"title":""},"id":"RCpXcZHoo0gc"},"source":["### Adding columns\n","To add a new column to your DataFrame, you can use the [withColumn](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withColumn.html) method:"]},{"cell_type":"code","execution_count":49,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"d526b302-1406-4f77-aa01-e083eead6489","showTitle":false,"title":""},"id":"DnEqzDIZpMft","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121887887,"user_tz":-480,"elapsed":12,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"75b51151-cf23-4333-e51e-17abd01eb6e7"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["[Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Romania', count=15, withinCountry=False),\n"," Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Croatia', count=1, withinCountry=False),\n"," Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Ireland', count=344, withinCountry=False),\n"," Row(DEST_COUNTRY_NAME='Egypt', ORIGIN_COUNTRY_NAME='United States', count=15, withinCountry=False),\n"," Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='India', count=62, withinCountry=False)]"]},"metadata":{},"execution_count":49}],"source":["flights_df.withColumn(\"withinCountry\", expr(\"ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME\"))\\\n",".take(5)"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"e8d282f2-124d-4145-a4b6-0395f98ba779","showTitle":false,"title":""},"id":"DGO9Qq5mp58F"},"source":["### Renaming columns\n","You can rename a column  with the [withColumnRenamed](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withColumnRenamed.html) method:"]},{"cell_type":"code","execution_count":50,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"9f00c2b9-63e0-4b78-b49b-908b0b568e5d","showTitle":false,"title":""},"id":"r1OA8jrjqSmI","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121888237,"user_tz":-480,"elapsed":359,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"3e0ca582-71f3-4438-fc48-958a2dc8e023"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------+-------------------+-----+\n","|  Destination|ORIGIN_COUNTRY_NAME|count|\n","+-------------+-------------------+-----+\n","|United States|            Romania|   15|\n","|United States|            Croatia|    1|\n","+-------------+-------------------+-----+\n","only showing top 2 rows\n","\n"]}],"source":["flights_df.withColumnRenamed(\"DEST_COUNTRY_NAME\", \"Destination\").show(2)"]},{"cell_type":"code","execution_count":51,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"dbf00d8f-8cc6-4371-adf5-d808c19f51cc","showTitle":false,"title":""},"id":"envGS7BvuZw-","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121888237,"user_tz":-480,"elapsed":8,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"2c228e2e-5752-488c-9b3c-bef0f9dac3f0"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------+-------+-----+\n","|         dest| origin|count|\n","+-------------+-------+-----+\n","|United States|Romania|   15|\n","|United States|Croatia|    1|\n","+-------------+-------+-----+\n","only showing top 2 rows\n","\n"]}],"source":["# renaming multiple columns\n","flights_df.withColumnRenamed(\"DEST_COUNTRY_NAME\", \"dest\")\\\n","  .withColumnRenamed(\"ORIGIN_COUNTRY_NAME\", \"origin\").show(2)"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"409332ac-160d-4a53-889a-a73be87b461a","showTitle":false,"title":""},"id":"wdJ4fEJjp8sn"},"source":["### Removing columns\n","[drop](https://spark.apache.org/docs/latest/api/python/reference/pyspark.pandas/api/pyspark.pandas.DataFrame.drop.html) is a dedicated method to remove columns from a DataFrame."]},{"cell_type":"code","execution_count":52,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"b6fd8952-cd1b-471d-84c8-db4d501ecd2e","showTitle":false,"title":""},"id":"PQjJwIIjyaZh","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121888742,"user_tz":-480,"elapsed":509,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"2c69222a-5809-4d4a-f365-8ae7082aa3c0"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["['DEST_COUNTRY_NAME', 'ORIGIN_COUNTRY_NAME']"]},"metadata":{},"execution_count":52}],"source":["flights_df.drop(\"count\").columns"]},{"cell_type":"code","execution_count":53,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"a39136dc-20d5-451a-868e-0324caba346b","showTitle":false,"title":""},"id":"KEl8Vz16zkLI","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121888743,"user_tz":-480,"elapsed":11,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"d7df95b8-d555-421e-8531-2e7fa16c2ff3"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+\n","|DEST_COUNTRY_NAME|\n","+-----------------+\n","|    United States|\n","|    United States|\n","|    United States|\n","|            Egypt|\n","|    United States|\n","+-----------------+\n","only showing top 5 rows\n","\n"]}],"source":["flights_df.drop(\"ORIGIN_COUNTRY_NAME\",\"count\").show(5)"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"8ad824ef-3c58-4ac4-97e5-23d9e3db1bb9","showTitle":false,"title":""},"id":"mxuu4VxOeL7E"},"source":["### Filtering Rows\n","There are two methods to perform filtering operations: you can use [where](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.where.html) or [filter](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.filter.html)\n","and they both will perform the same operation and accept the same argument types when used\n","with DataFrames. To filter rows, you need an expression that evaluates to true or false."]},{"cell_type":"code","execution_count":54,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"c6efd06e-5afd-43e4-b3db-03da380c7244","showTitle":false,"title":""},"id":"jqOEwTWb0Cuf","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121889444,"user_tz":-480,"elapsed":706,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"32ac993a-90d3-4c98-d0b7-f66282105919"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-------------------+-----+\n","|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n","+-----------------+-------------------+-----+\n","|    United States|            Croatia|    1|\n","|    United States|          Singapore|    1|\n","+-----------------+-------------------+-----+\n","only showing top 2 rows\n","\n"]}],"source":["flights_df.filter(col(\"count\") < 2).show(2)"]},{"cell_type":"code","execution_count":55,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"78f57089-ef38-4886-9f76-2eae587eb019","showTitle":false,"title":""},"id":"LlgpafgU14kN","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121889444,"user_tz":-480,"elapsed":6,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"4a18b22b-2d05-4300-b7aa-66f5a663769e"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-------------------+-----+\n","|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n","+-----------------+-------------------+-----+\n","|    United States|            Croatia|    1|\n","|    United States|          Singapore|    1|\n","+-----------------+-------------------+-----+\n","only showing top 2 rows\n","\n"]}],"source":["flights_df.where(\"count < 2\").show(2)"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"38e32219-14a9-44a7-8b59-3d587008fa9c","showTitle":false,"title":""},"id":"hKIb6DuC28jS"},"source":["You might want to put multiple filters into the same expression, but this is not always useful, because Spark automatically performs all filtering operations at\n","the same time regardless of the filter ordering.\n","If you want to specify multiple filters, just chain them sequentially and let Spark handle the rest."]},{"cell_type":"code","execution_count":56,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"1b07de48-6691-46c0-b5e8-aad1502d131b","showTitle":false,"title":""},"id":"_qfyk5lg3R6z","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121889843,"user_tz":-480,"elapsed":402,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"22f83ccd-b964-456b-f975-0477f081a205"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-------------------+-----+\n","|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n","+-----------------+-------------------+-----+\n","|    United States|            Croatia|    1|\n","|          Moldova|      United States|    1|\n","+-----------------+-------------------+-----+\n","only showing top 2 rows\n","\n"]}],"source":["flights_df.where(col(\"count\") < 2).where(col(\"ORIGIN_COUNTRY_NAME\") != \"Singapore\")\\\n",".show(2)"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"8bdb8b45-2c48-47af-9162-de9dadc10192","showTitle":false,"title":""},"id":"R6WFGhJUeaW4"},"source":["### Getting Unique Rows\n","To extract the unique or distinct values in a DataFrame, you can use the [distinct](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.distinct.html?highlight=distinct#pyspark.sql.DataFrame.distinct) method on a\n","DataFrame."]},{"cell_type":"code","execution_count":57,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"c09d4b13-dcb9-4ef9-b4c3-c9438e9472e7","showTitle":false,"title":""},"id":"JGK58YPC4Tja","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121891264,"user_tz":-480,"elapsed":1424,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"a68e0c30-6e2d-4173-f98d-b06bbaa82ce1"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["256"]},"metadata":{},"execution_count":57}],"source":["flights_df.select(\"ORIGIN_COUNTRY_NAME\", \"DEST_COUNTRY_NAME\").distinct().count()"]},{"cell_type":"code","execution_count":58,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"a46d85a0-3648-44c9-a94f-e495baaec136","showTitle":false,"title":""},"id":"Ja-q7QEd5fcJ","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121892408,"user_tz":-480,"elapsed":1148,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"ea3f92bc-617e-4156-bea9-3050de0c7ed9"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["125"]},"metadata":{},"execution_count":58}],"source":["flights_df.select(\"ORIGIN_COUNTRY_NAME\").distinct().count()"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"5d66d6ef-48d3-4ca5-9142-aa535c41a49f","showTitle":false,"title":""},"id":"dcv4fmAkegK2"},"source":["### Random Samples\n","To sample some random records from your DataFrame, use the [sample](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.sample.html) method."]},{"cell_type":"code","execution_count":59,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"39e901ef-411c-4640-af26-4a7510570825","showTitle":false,"title":""},"id":"T0YcaJcE6jHf","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121892866,"user_tz":-480,"elapsed":461,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"e2dc95ee-7664-45ad-a813-199c74fba82b"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["138"]},"metadata":{},"execution_count":59}],"source":["flights_df.sample(withReplacement = False,\n","                      fraction= 0.5,\n","                      seed = 5).count()"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"15e9b5fb-064a-4f7d-bc2e-7edf8d46d3f0","showTitle":false,"title":""},"id":"cx5Zh0ikeihg"},"source":["### Random Splits\n","You may need to break up your DataFrame into random splits to use with machine learning algorithms to create training,\n","validation, and test sets. You can do this using the [randomSplit](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.randomSplit.html) method."]},{"cell_type":"code","execution_count":60,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"521c4f24-1cc5-438e-aefe-010abc9b9163","showTitle":false,"title":""},"id":"VG_mZQlD7n7T","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121893343,"user_tz":-480,"elapsed":481,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"ede12c01-597a-4d6e-dac6-141f90a49665"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["[DataFrame[DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string, count: int],\n"," DataFrame[DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string, count: int]]"]},"metadata":{},"execution_count":60}],"source":["flightDataSplits = flights_df.randomSplit([0.25, 0.75], seed = 5)\n","\n","flightDataSplits"]},{"cell_type":"code","execution_count":61,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"1d2479cd-f9c9-4232-9838-c855468a5b8d","showTitle":false,"title":""},"id":"XLVwUug78QqG","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121894577,"user_tz":-480,"elapsed":1237,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"a2c2b488-6aad-4da2-ddd7-dc33229f877f"},"outputs":[{"output_type":"execute_result","data":{"text/plain":["True"]},"metadata":{},"execution_count":61}],"source":["flightDataSplits[0].count() < flightDataSplits[1].count()"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"ebe66cc0-8e84-43be-a169-5b00ed2f0196","showTitle":false,"title":""},"id":"-eAiauqNe0qY"},"source":["### Sorting Rows\n","\n","There are two equivalent operations to sort the values in a DataFrame: [sort](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.sort.html) and [orderBy](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.orderBy.html)."]},{"cell_type":"code","execution_count":62,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"8cc7f451-2b88-413b-9b53-22c96b9c3df0","showTitle":false,"title":""},"id":"MuEY7GSXCgPh","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121894979,"user_tz":-480,"elapsed":404,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"b56eefaf-9a98-4ac5-d5d1-498ecdddd07b"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-------------------+------+\n","|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME| count|\n","+-----------------+-------------------+------+\n","|    United States|      United States|370002|\n","|    United States|             Canada|  8483|\n","|           Canada|      United States|  8399|\n","|    United States|             Mexico|  7187|\n","|           Mexico|      United States|  7140|\n","+-----------------+-------------------+------+\n","only showing top 5 rows\n","\n"]}],"source":["flights_df.sort(\"count\", ascending=False).show(5)"]},{"cell_type":"code","execution_count":63,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"c36d6ef7-78c3-4fa8-bda5-a44bcbac30b4","showTitle":false,"title":""},"id":"0GZxK726Cn_P","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121894980,"user_tz":-480,"elapsed":8,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"65871ccf-38f8-423d-8c61-48abe6114183"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-------------------+-----+\n","|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n","+-----------------+-------------------+-----+\n","|     Burkina Faso|      United States|    1|\n","|    Cote d'Ivoire|      United States|    1|\n","|           Cyprus|      United States|    1|\n","|         Djibouti|      United States|    1|\n","|        Indonesia|      United States|    1|\n","+-----------------+-------------------+-----+\n","only showing top 5 rows\n","\n"]}],"source":["flights_df.orderBy(\"count\", \"DEST_COUNTRY_NAME\").show(5)"]},{"cell_type":"code","execution_count":64,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"117e51cc-7aa4-4bca-884e-84e55cb7d067","showTitle":false,"title":""},"id":"FGhto_vKCqGO","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121895375,"user_tz":-480,"elapsed":400,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"fffadf61-1de4-45c0-e98f-0acd9a341b8f"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-------------------+------+\n","|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME| count|\n","+-----------------+-------------------+------+\n","|    United States|      United States|370002|\n","|    United States|             Canada|  8483|\n","|           Canada|      United States|  8399|\n","|    United States|             Mexico|  7187|\n","|           Mexico|      United States|  7140|\n","+-----------------+-------------------+------+\n","only showing top 5 rows\n","\n"]}],"source":["flights_df.orderBy(col(\"count\"), col(\"DEST_COUNTRY_NAME\"), ascending=False).show(5)"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"287469bf-e405-45f0-8190-3c54d5ba8b8c","showTitle":false,"title":""},"id":"FGZdgNbqjXZs"},"source":["Let’s find the top five destination countries in the data."]},{"cell_type":"code","execution_count":65,"metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"85bc860c-eeca-47e4-87a0-9d356e804aee","showTitle":false,"title":""},"id":"hvvsFAKAjXZs","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121896242,"user_tz":-480,"elapsed":871,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"284aeb9a-bdc1-4380-b7a4-2c884a1e1406"},"outputs":[{"output_type":"stream","name":"stdout","text":["+-----------------+-----------------+\n","|DEST_COUNTRY_NAME|destination_total|\n","+-----------------+-----------------+\n","|    United States|           411352|\n","|           Canada|             8399|\n","|           Mexico|             7140|\n","|   United Kingdom|             2025|\n","|            Japan|             1548|\n","+-----------------+-----------------+\n","\n"]}],"source":["from pyspark.sql.functions import desc\n","\n","flights_df\\\n","  .groupBy(\"DEST_COUNTRY_NAME\")\\\n","  .sum(\"count\")\\\n","  .withColumnRenamed(\"sum(count)\", \"destination_total\")\\\n","  .sort(desc(\"destination_total\"))\\\n","  .limit(5)\\\n","  .show()"]},{"cell_type":"markdown","metadata":{"application/vnd.databricks.v1+cell":{"cellMetadata":{"byteLimit":2048000,"rowLimit":10000},"inputWidgets":{},"nuid":"035fd802-f3cc-4324-af50-5d1ce90f22b7","showTitle":false,"title":""},"id":"Ycn94bvvESx9"},"source":["You can also sort within each partition using the [sortWithinPartitions](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.sortWithinPartitions.html) method."]},{"cell_type":"markdown","source":["# Spark MLlib\n","\n","* [pyspark.ml](https://spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html): DataFrame-based Machine Learning library.\n","\n","* It provides APIs for exploratory data analysis, feature engineering, model training, model evaluation, tuning, pipelines, statisitics, linear algebra.\n","\n","* [Spark Machine Learning Library (MLlib) Guide](https://spark.apache.org/docs/latest/ml-guide.html).\n","\n","* [pyspark.ml.linalg](https://spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html#vector-and-matrix): data types for representing vectors and matrics.\n","* [pyspark.ml.stat](https://spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html#statistics): Summarizer, Correlation, and ChiSquareTest.\n","\n","* [pyspark.ml.Transformer](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.Transformer.html): The abstract class for transofmers.\n","\n","\n","* [pyspark.ml.Estimator](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.Estimator.html): The abstract class for estimators.\n","\n","* [pyspark.ml.feature](https://spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html#feature): transformers and estimators for feature extraction, transformation, and selection.\n"],"metadata":{"id":"BFR568Ug3knV"}},{"cell_type":"markdown","source":["\n","In this section, we only use small training data to explore how to use the MLlib API to train machine learning models."],"metadata":{"id":"SlcGJH7RenyM"}},{"cell_type":"markdown","source":["## Classification\n","* `pyspark.ml.classification`: [PySpark API: Classification](https://spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html#classification)\n","* [Spark Programming Guide: Classification](https://spark.apache.org/docs/latest/ml-classification-regression.html#classification)"],"metadata":{"id":"oIS2aXl6FtSA"}},{"cell_type":"markdown","source":["## Dataset\n"],"metadata":{"id":"wwERJtwyw0uI"}},{"cell_type":"code","source":["bInputPath = \"/content/SDSC-Spark4/data/binary_class\""],"metadata":{"id":"HBGTUg91GB8S","executionInfo":{"status":"ok","timestamp":1692121896243,"user_tz":-480,"elapsed":4,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":66,"outputs":[]},{"cell_type":"code","source":["bInput = spark.read.parquet(bInputPath).selectExpr(\"features\", \"cast(label as double) as label\")"],"metadata":{"id":"WyAQaKexcVj9","executionInfo":{"status":"ok","timestamp":1692121896603,"user_tz":-480,"elapsed":363,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":67,"outputs":[]},{"cell_type":"code","source":["bInput.printSchema()"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"zeMCH_fuNcKT","executionInfo":{"status":"ok","timestamp":1692121896604,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"b6a641f0-df50-4fbc-b86c-78aed05129c7"},"execution_count":68,"outputs":[{"output_type":"stream","name":"stdout","text":["root\n"," |-- features: vector (nullable = true)\n"," |-- label: double (nullable = true)\n","\n"]}]},{"cell_type":"code","source":["bInput.show(2)"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"XdrlHpPdMKlm","executionInfo":{"status":"ok","timestamp":1692121897107,"user_tz":-480,"elapsed":507,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"6a6be324-930a-4e79-9450-4ba3fc2f7606"},"execution_count":69,"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------+-----+\n","|     features|label|\n","+-------------+-----+\n","|[0.0,0.0,0.0]|  1.0|\n","|[0.0,0.0,0.0]|  0.0|\n","+-------------+-----+\n","only showing top 2 rows\n","\n"]}]},{"cell_type":"markdown","source":["## Logistic Regression\n","* [LogisticRegression](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.LogisticRegression.html) is used to predict a binary outcome.\n","\n","* [LogisticRegressionModel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.LogisticRegressionModel.html): Model fitted by LogisticRegression."],"metadata":{"id":"9TrUvOFzGFQ5"}},{"cell_type":"code","source":["from pyspark.ml.classification import LogisticRegression\n","\n","lr = LogisticRegression()"],"metadata":{"id":"BRAAS20wL2NU","executionInfo":{"status":"ok","timestamp":1692121897495,"user_tz":-480,"elapsed":390,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":70,"outputs":[]},{"cell_type":"markdown","source":["### Parameters"],"metadata":{"id":"ux_oJcBVP-Wu"}},{"cell_type":"code","source":["print(lr.explainParams())"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"IRxJGVFvQAoi","executionInfo":{"status":"ok","timestamp":1692121898181,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"699e8c57-548f-4d39-9f04-b7cf5f89ff72"},"execution_count":71,"outputs":[{"output_type":"stream","name":"stdout","text":["aggregationDepth: suggested depth for treeAggregate (>= 2). (default: 2)\n","elasticNetParam: the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty. (default: 0.0)\n","family: The name of family which is a description of the label distribution to be used in the model. Supported options: auto, binomial, multinomial (default: auto)\n","featuresCol: features column name. (default: features)\n","fitIntercept: whether to fit an intercept term. (default: True)\n","labelCol: label column name. (default: label)\n","lowerBoundsOnCoefficients: The lower bounds on coefficients if fitting under bound constrained optimization. The bound matrix must be compatible with the shape (1, number of features) for binomial regression, or (number of classes, number of features) for multinomial regression. (undefined)\n","lowerBoundsOnIntercepts: The lower bounds on intercepts if fitting under bound constrained optimization. The bounds vector size must beequal with 1 for binomial regression, or the number oflasses for multinomial regression. (undefined)\n","maxBlockSizeInMB: maximum memory in MB for stacking input data into blocks. Data is stacked within partitions. If more than remaining data size in a partition then it is adjusted to the data size. Default 0.0 represents choosing optimal value, depends on specific algorithm. Must be >= 0. (default: 0.0)\n","maxIter: max number of iterations (>= 0). (default: 100)\n","predictionCol: prediction column name. (default: prediction)\n","probabilityCol: Column name for predicted class conditional probabilities. Note: Not all models output well-calibrated probability estimates! These probabilities should be treated as confidences, not precise probabilities. (default: probability)\n","rawPredictionCol: raw prediction (a.k.a. confidence) column name. (default: rawPrediction)\n","regParam: regularization parameter (>= 0). (default: 0.0)\n","standardization: whether to standardize the training features before fitting the model. (default: True)\n","threshold: Threshold in binary classification prediction, in range [0, 1]. If threshold and thresholds are both set, they must match.e.g. if threshold is p, then thresholds must be equal to [1-p, p]. (default: 0.5)\n","thresholds: Thresholds in multi-class classification to adjust the probability of predicting each class. Array must have length equal to the number of classes, with values > 0, excepting that at most one value may be 0. The class with largest value p/t is predicted, where p is the original probability of that class and t is the class's threshold. (undefined)\n","tol: the convergence tolerance for iterative algorithms (>= 0). (default: 1e-06)\n","upperBoundsOnCoefficients: The upper bounds on coefficients if fitting under bound constrained optimization. The bound matrix must be compatible with the shape (1, number of features) for binomial regression, or (number of classes, number of features) for multinomial regression. (undefined)\n","upperBoundsOnIntercepts: The upper bounds on intercepts if fitting under bound constrained optimization. The bound vector size must be equal with 1 for binomial regression, or the number of classes for multinomial regression. (undefined)\n","weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0. (undefined)\n"]}]},{"cell_type":"markdown","source":["### Fit a model"],"metadata":{"id":"FzEytCDVkBo_"}},{"cell_type":"markdown","source":["You can train the model with the [LogisticRegression.fit()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.LogisticRegression.html#pyspark.ml.classification.LogisticRegression.fit) method and get a [LogisticRegressionModel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.LogisticRegressionModel.html)."],"metadata":{"id":"aywExzKexnaT"}},{"cell_type":"code","source":["lrModel = lr.fit(bInput)"],"metadata":{"id":"cHaZ130AMBRR","executionInfo":{"status":"ok","timestamp":1692121908454,"user_tz":-480,"elapsed":10278,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":72,"outputs":[]},{"cell_type":"code","source":["lrModel.coefficients"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"CVG1nfRYRZqd","executionInfo":{"status":"ok","timestamp":1692121908454,"user_tz":-480,"elapsed":36,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"60bf19c5-a472-4cea-d1b4-db3a83a1fe92"},"execution_count":73,"outputs":[{"output_type":"execute_result","data":{"text/plain":["DenseVector([18.7224, -0.5694, 9.3612])"]},"metadata":{},"execution_count":73}]},{"cell_type":"code","source":["lrModel.intercept"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"SsXPs_NeRjj-","executionInfo":{"status":"ok","timestamp":1692121908455,"user_tz":-480,"elapsed":31,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"3307f25a-532e-4d46-db85-b5d5b46c63a1"},"execution_count":74,"outputs":[{"output_type":"execute_result","data":{"text/plain":["-28.04329511868945"]},"metadata":{},"execution_count":74}]},{"cell_type":"markdown","source":["### Training Summary"],"metadata":{"id":"wJAxvGjaWeve"}},{"cell_type":"markdown","source":["[BinaryLogisticRegressionTrainingSummary](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.BinaryLogisticRegressionTrainingSummary.html)  provides a summary for a binary LogisticRegressionModel."],"metadata":{"id":"Onf6cb40ThE-"}},{"cell_type":"code","source":["lrSummary = lrModel.summary\n","lrSummary"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"g63i_SeMMb1j","executionInfo":{"status":"ok","timestamp":1692121908456,"user_tz":-480,"elapsed":28,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"0ff535db-6b7c-421b-ea50-b622fe6bc34a"},"execution_count":75,"outputs":[{"output_type":"execute_result","data":{"text/plain":["<pyspark.ml.classification.BinaryLogisticRegressionTrainingSummary at 0x7acc41a73430>"]},"metadata":{},"execution_count":75}]},{"cell_type":"code","source":["lrSummary.areaUnderROC"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"JRtRArS9SRjt","executionInfo":{"status":"ok","timestamp":1692121908966,"user_tz":-480,"elapsed":535,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"ad171ba3-4c9e-408a-cea9-9840e7d3d33e"},"execution_count":76,"outputs":[{"output_type":"execute_result","data":{"text/plain":["0.6666666666666666"]},"metadata":{},"execution_count":76}]},{"cell_type":"code","source":["lrSummary.objectiveHistory"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"7VPwv8cwM3lp","executionInfo":{"status":"ok","timestamp":1692121908966,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"d70bc9c6-d182-422d-989d-0458f01c82c1"},"execution_count":77,"outputs":[{"output_type":"execute_result","data":{"text/plain":["[0.6730116670092563,\n"," 0.30533476678669746,\n"," 0.19572951692227342,\n"," 0.08238560717506734,\n"," 0.039904390712412516,\n"," 0.019187605729977825,\n"," 0.009480513129879626,\n"," 0.004700793975398925,\n"," 0.002342824005088814,\n"," 0.0011692212872630964,\n"," 0.0005841333526453686,\n"," 0.00029193843681446,\n"," 0.00014593757317782447,\n"," 7.295887614374282e-05,\n"," 3.647309882223227e-05,\n"," 1.822801708342421e-05,\n"," 9.095755464927005e-06,\n"," 4.50530629284565e-06,\n"," 2.1743484095163617e-06,\n"," 1.0422594942126269e-06,\n"," 5.280808738948462e-07,\n"," 2.628531186444535e-07,\n"," 1.3166032239693124e-07,\n"," 6.578498712560823e-08,\n"," 3.290121373800775e-08,\n"," 1.6448921648781483e-08,\n"," 8.224786126080745e-09]"]},"metadata":{},"execution_count":77}]},{"cell_type":"markdown","source":["##Decision Tree Classifier\n","\n","* A Decision Tree is a set of if-then-else rules learned from the training data\n","* [Spark Programming Guide: Decision Trees](https://spark.apache.org/docs/latest/ml-classification-regression.html#decision-trees)\n","\n","* [pyspark.ml.classification.DecisionTreeClassifier](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.DecisionTreeClassifier.html)\n","\n","* [DecisionTreeClassificationModel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.DecisionTreeClassificationModel.html)\n"],"metadata":{"id":"pvAo5F1MQAgc"}},{"cell_type":"code","source":["from pyspark.ml.classification import DecisionTreeClassifier\n","\n","dt = DecisionTreeClassifier()"],"metadata":{"id":"MWICHqVlQOrK","executionInfo":{"status":"ok","timestamp":1692121909367,"user_tz":-480,"elapsed":406,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":78,"outputs":[]},{"cell_type":"markdown","source":["### Parameters"],"metadata":{"id":"YW3Hbn1MZ-27"}},{"cell_type":"code","source":["print(dt.explainParams())"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"5yAXOh3UaAz_","executionInfo":{"status":"ok","timestamp":1692121909367,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"b608b044-0307-48cc-9604-d6ea30291c88"},"execution_count":79,"outputs":[{"output_type":"stream","name":"stdout","text":["cacheNodeIds: If false, the algorithm will pass trees to executors to match instances with nodes. If true, the algorithm will cache node IDs for each instance. Caching can speed up training of deeper trees. Users can set how often should the cache be checkpointed or disable it by setting checkpointInterval. (default: False)\n","checkpointInterval: set checkpoint interval (>= 1) or disable checkpoint (-1). E.g. 10 means that the cache will get checkpointed every 10 iterations. Note: this setting will be ignored if the checkpoint directory is not set in the SparkContext. (default: 10)\n","featuresCol: features column name. (default: features)\n","impurity: Criterion used for information gain calculation (case-insensitive). Supported options: entropy, gini (default: gini)\n","labelCol: label column name. (default: label)\n","leafCol: Leaf indices column name. Predicted leaf index of each instance in each tree by preorder. (default: )\n","maxBins: Max number of bins for discretizing continuous features.  Must be >=2 and >= number of categories for any categorical feature. (default: 32)\n","maxDepth: Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30]. (default: 5)\n","maxMemoryInMB: Maximum memory in MB allocated to histogram aggregation. If too small, then 1 node will be split per iteration, and its aggregates may exceed this size. (default: 256)\n","minInfoGain: Minimum information gain for a split to be considered at a tree node. (default: 0.0)\n","minInstancesPerNode: Minimum number of instances each child must have after split. If a split causes the left or right child to have fewer than minInstancesPerNode, the split will be discarded as invalid. Should be >= 1. (default: 1)\n","minWeightFractionPerNode: Minimum fraction of the weighted sample count that each child must have after split. If a split causes the fraction of the total weight in the left or right child to be less than minWeightFractionPerNode, the split will be discarded as invalid. Should be in interval [0.0, 0.5). (default: 0.0)\n","predictionCol: prediction column name. (default: prediction)\n","probabilityCol: Column name for predicted class conditional probabilities. Note: Not all models output well-calibrated probability estimates! These probabilities should be treated as confidences, not precise probabilities. (default: probability)\n","rawPredictionCol: raw prediction (a.k.a. confidence) column name. (default: rawPrediction)\n","seed: random seed. (default: 8134934633765591589)\n","thresholds: Thresholds in multi-class classification to adjust the probability of predicting each class. Array must have length equal to the number of classes, with values > 0, excepting that at most one value may be 0. The class with largest value p/t is predicted, where p is the original probability of that class and t is the class's threshold. (undefined)\n","weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0. (undefined)\n"]}]},{"cell_type":"markdown","source":["### Fit a model"],"metadata":{"id":"ufp-kKCGkc4j"}},{"cell_type":"code","source":["dtModel = dt.fit(bInput)"],"metadata":{"id":"afKv66dhQT1T","executionInfo":{"status":"ok","timestamp":1692121912013,"user_tz":-480,"elapsed":2652,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":80,"outputs":[]},{"cell_type":"markdown","source":["### Feature Importance\n","You can extract the feature importance scores from your model using [featureImportances](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.DecisionTreeClassificationModel.html#pyspark.ml.classification.DecisionTreeClassificationModel.featureImportances):"],"metadata":{"id":"6DzwcsImd1cd"}},{"cell_type":"code","source":["dtModel.featureImportances"],"metadata":{"id":"qj0GbMCDdyRh","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121912015,"user_tz":-480,"elapsed":34,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"e9f29fe5-764d-49c4-ee15-620a7788db96"},"execution_count":81,"outputs":[{"output_type":"execute_result","data":{"text/plain":["SparseVector(3, {0: 1.0})"]},"metadata":{},"execution_count":81}]},{"cell_type":"markdown","source":["## Random Forest Classifier\n","* Random forests are ensembles of decision trees (combine many decision trees in order to reduce the risk of overfitting).\n","* [RandomForestClassifier](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.RandomForestClassifier.html)\n","* [RandomForestClassificationModel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.RandomForestClassificationModel.html)"],"metadata":{"id":"aNBNV4EhQlQi"}},{"cell_type":"code","source":["from pyspark.ml.classification import RandomForestClassifier\n","\n","rfClassifier = RandomForestClassifier()"],"metadata":{"id":"aPzQZkeRQxmk","executionInfo":{"status":"ok","timestamp":1692121912016,"user_tz":-480,"elapsed":28,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":82,"outputs":[]},{"cell_type":"markdown","source":["### Parameters"],"metadata":{"id":"R_KbS8MvbYwU"}},{"cell_type":"code","source":["print(rfClassifier.explainParams())"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"3NAXTMPHbbKH","executionInfo":{"status":"ok","timestamp":1692121912017,"user_tz":-480,"elapsed":29,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"a12256ac-0308-43a7-ccf0-de1e28d2fbfc"},"execution_count":83,"outputs":[{"output_type":"stream","name":"stdout","text":["bootstrap: Whether bootstrap samples are used when building trees. (default: True)\n","cacheNodeIds: If false, the algorithm will pass trees to executors to match instances with nodes. If true, the algorithm will cache node IDs for each instance. Caching can speed up training of deeper trees. Users can set how often should the cache be checkpointed or disable it by setting checkpointInterval. (default: False)\n","checkpointInterval: set checkpoint interval (>= 1) or disable checkpoint (-1). E.g. 10 means that the cache will get checkpointed every 10 iterations. Note: this setting will be ignored if the checkpoint directory is not set in the SparkContext. (default: 10)\n","featureSubsetStrategy: The number of features to consider for splits at each tree node. Supported options: 'auto' (choose automatically for task: If numTrees == 1, set to 'all'. If numTrees > 1 (forest), set to 'sqrt' for classification and to 'onethird' for regression), 'all' (use all features), 'onethird' (use 1/3 of the features), 'sqrt' (use sqrt(number of features)), 'log2' (use log2(number of features)), 'n' (when n is in the range (0, 1.0], use n * number of features. When n is in the range (1, number of features), use n features). default = 'auto' (default: auto)\n","featuresCol: features column name. (default: features)\n","impurity: Criterion used for information gain calculation (case-insensitive). Supported options: entropy, gini (default: gini)\n","labelCol: label column name. (default: label)\n","leafCol: Leaf indices column name. Predicted leaf index of each instance in each tree by preorder. (default: )\n","maxBins: Max number of bins for discretizing continuous features.  Must be >=2 and >= number of categories for any categorical feature. (default: 32)\n","maxDepth: Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30]. (default: 5)\n","maxMemoryInMB: Maximum memory in MB allocated to histogram aggregation. If too small, then 1 node will be split per iteration, and its aggregates may exceed this size. (default: 256)\n","minInfoGain: Minimum information gain for a split to be considered at a tree node. (default: 0.0)\n","minInstancesPerNode: Minimum number of instances each child must have after split. If a split causes the left or right child to have fewer than minInstancesPerNode, the split will be discarded as invalid. Should be >= 1. (default: 1)\n","minWeightFractionPerNode: Minimum fraction of the weighted sample count that each child must have after split. If a split causes the fraction of the total weight in the left or right child to be less than minWeightFractionPerNode, the split will be discarded as invalid. Should be in interval [0.0, 0.5). (default: 0.0)\n","numTrees: Number of trees to train (>= 1). (default: 20)\n","predictionCol: prediction column name. (default: prediction)\n","probabilityCol: Column name for predicted class conditional probabilities. Note: Not all models output well-calibrated probability estimates! These probabilities should be treated as confidences, not precise probabilities. (default: probability)\n","rawPredictionCol: raw prediction (a.k.a. confidence) column name. (default: rawPrediction)\n","seed: random seed. (default: -5518030150412635304)\n","subsamplingRate: Fraction of the training data used for learning each decision tree, in range (0, 1]. (default: 1.0)\n","thresholds: Thresholds in multi-class classification to adjust the probability of predicting each class. Array must have length equal to the number of classes, with values > 0, excepting that at most one value may be 0. The class with largest value p/t is predicted, where p is the original probability of that class and t is the class's threshold. (undefined)\n","weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0. (undefined)\n"]}]},{"cell_type":"markdown","source":["* [numTrees](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.RandomForestClassifier.html#pyspark.ml.classification.RandomForestClassifier.numTrees)\n","\n","* [featureSubsetStrategy](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.RandomForestClassifier.html#pyspark.ml.classification.RandomForestClassifier.featureSubsetStrategy)\n","\n","* [subsamplingRate](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.RandomForestClassifier.html#pyspark.ml.classification.RandomForestClassifier.subsamplingRate)"],"metadata":{"id":"0MeU6VQ8bs3T"}},{"cell_type":"code","source":["rfClassifier.getNumTrees()"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"GqopigSCdipi","executionInfo":{"status":"ok","timestamp":1692121912018,"user_tz":-480,"elapsed":25,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"db58255f-40b1-4d56-fcfe-e1cb93506648"},"execution_count":84,"outputs":[{"output_type":"execute_result","data":{"text/plain":["20"]},"metadata":{},"execution_count":84}]},{"cell_type":"code","source":["rfClassifier.getFeatureSubsetStrategy()"],"metadata":{"colab":{"base_uri":"https://localhost:8080/","height":35},"id":"SoDhDaUgdtz7","executionInfo":{"status":"ok","timestamp":1692121912018,"user_tz":-480,"elapsed":21,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"96a93e71-83a4-40c4-cc07-434297192cc3"},"execution_count":85,"outputs":[{"output_type":"execute_result","data":{"text/plain":["'auto'"],"application/vnd.google.colaboratory.intrinsic+json":{"type":"string"}},"metadata":{},"execution_count":85}]},{"cell_type":"code","source":["rfClassifier.getSubsamplingRate()"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"RYQrWq9rd07Q","executionInfo":{"status":"ok","timestamp":1692121912019,"user_tz":-480,"elapsed":21,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"302bd8c1-0def-4026-abb3-7ec7c7271b5c"},"execution_count":86,"outputs":[{"output_type":"execute_result","data":{"text/plain":["1.0"]},"metadata":{},"execution_count":86}]},{"cell_type":"markdown","source":["### Fit a model"],"metadata":{"id":"WG0AMpbtbcjo"}},{"cell_type":"code","source":["rfModel = rfClassifier.fit(bInput)"],"metadata":{"id":"CU-EQP_3Q4rm","executionInfo":{"status":"ok","timestamp":1692121914311,"user_tz":-480,"elapsed":2309,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":87,"outputs":[]},{"cell_type":"markdown","source":["### Training Summary\n"],"metadata":{"id":"U8b8M5gccYQA"}},{"cell_type":"code","source":["rfSummary = rfModel.summary\n","rfSummary"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"FIUcM1b_gYnX","executionInfo":{"status":"ok","timestamp":1692121914312,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"3bbb2339-3d89-48ab-f1fe-7444923d7c79"},"execution_count":88,"outputs":[{"output_type":"execute_result","data":{"text/plain":["<pyspark.ml.classification.BinaryRandomForestClassificationTrainingSummary at 0x7acc40099f90>"]},"metadata":{},"execution_count":88}]},{"cell_type":"markdown","source":["[BinaryRandomForestClassificationTrainingSummary](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.BinaryRandomForestClassificationTrainingSummary.html)"],"metadata":{"id":"bRTcyWvPg8Pt"}},{"cell_type":"code","source":["rfSummary.falsePositiveRateByLabel"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"oKz4sMW9gox2","executionInfo":{"status":"ok","timestamp":1692121914855,"user_tz":-480,"elapsed":549,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"bcad145c-6eb3-4a23-e61a-b38f7d28ae74"},"execution_count":89,"outputs":[{"output_type":"execute_result","data":{"text/plain":["[0.6666666666666666, 0.0]"]},"metadata":{},"execution_count":89}]},{"cell_type":"markdown","source":["## Other classification algorithms in MLlib\n","* Gradient Boosted Trees: [GBTClassifier](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.GBTClassifier.html)\n","* Naive Bayes: [NaiveBayes](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.classification.NaiveBayes.html)"],"metadata":{"id":"0CGgsl3QQ5g6"}},{"cell_type":"markdown","source":["#Regression\n","\n","* `pyspark.ml.regression`: [PySpark API: Regression](https://spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html#regression)\n","* [Spark Programming Guide: Regression](https://spark.apache.org/docs/latest/ml-classification-regression.html#regression)"],"metadata":{"id":"-NuWrHpngg21"}},{"cell_type":"markdown","source":["## Dataset"],"metadata":{"id":"VnqvYUL7xtl-"}},{"cell_type":"code","source":["regPath = \"/content/SDSC-Spark4/data/regression\""],"metadata":{"id":"6KSwilLjfit5","executionInfo":{"status":"ok","timestamp":1692121914856,"user_tz":-480,"elapsed":16,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":90,"outputs":[]},{"cell_type":"code","source":["regDF = spark.read.parquet(regPath)\n","regDF.printSchema()"],"metadata":{"id":"-bpJPHVuofic","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121914857,"user_tz":-480,"elapsed":16,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"55a10865-4f72-47cf-bb16-cf852596796b"},"execution_count":91,"outputs":[{"output_type":"stream","name":"stdout","text":["root\n"," |-- features: vector (nullable = true)\n"," |-- label: double (nullable = true)\n","\n"]}]},{"cell_type":"code","source":["regDF.show(2)"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"HlfE66sXEefc","executionInfo":{"status":"ok","timestamp":1692121914857,"user_tz":-480,"elapsed":12,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"402a9237-8607-4d70-dfc7-07b314662da7"},"execution_count":92,"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------+-----+\n","|     features|label|\n","+-------------+-----+\n","|[0.0,0.0,0.0]|  2.0|\n","|[0.0,0.0,0.0]|  1.0|\n","+-------------+-----+\n","only showing top 2 rows\n","\n"]}]},{"cell_type":"code","source":["regDF.write.parquet(\"/content/drive/MyDrive/sample_data/regression\")"],"metadata":{"id":"nd1CjJbENoc_","executionInfo":{"status":"ok","timestamp":1692121915590,"user_tz":-480,"elapsed":741,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":93,"outputs":[]},{"cell_type":"markdown","source":["## Linear Regression\n","* [LinearRegression](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.LinearRegression.html) is used to predict a real number from a set of numeric features.\n","* [LinearRegressionModel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.LinearRegressionModel.html)"],"metadata":{"id":"UGC-jlBEfZxf"}},{"cell_type":"code","source":["from pyspark.ml.regression import LinearRegression\n","\n","linear = LinearRegression().setMaxIter(10).setRegParam(0.3).setElasticNetParam(0.8)"],"metadata":{"id":"I7mkhNvSfjcg","executionInfo":{"status":"ok","timestamp":1692121915590,"user_tz":-480,"elapsed":8,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":94,"outputs":[]},{"cell_type":"markdown","source":["### Parameters"],"metadata":{"id":"jvZTMNh7puik"}},{"cell_type":"code","source":["print(linear.explainParams())"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"CKOYNerfpvCU","executionInfo":{"status":"ok","timestamp":1692121915591,"user_tz":-480,"elapsed":8,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"3b34a346-ac28-44ef-d73c-bd611c3a3f95"},"execution_count":95,"outputs":[{"output_type":"stream","name":"stdout","text":["aggregationDepth: suggested depth for treeAggregate (>= 2). (default: 2)\n","elasticNetParam: the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty. (default: 0.0, current: 0.8)\n","epsilon: The shape parameter to control the amount of robustness. Must be > 1.0. Only valid when loss is huber (default: 1.35)\n","featuresCol: features column name. (default: features)\n","fitIntercept: whether to fit an intercept term. (default: True)\n","labelCol: label column name. (default: label)\n","loss: The loss function to be optimized. Supported options: squaredError, huber. (default: squaredError)\n","maxBlockSizeInMB: maximum memory in MB for stacking input data into blocks. Data is stacked within partitions. If more than remaining data size in a partition then it is adjusted to the data size. Default 0.0 represents choosing optimal value, depends on specific algorithm. Must be >= 0. (default: 0.0)\n","maxIter: max number of iterations (>= 0). (default: 100, current: 10)\n","predictionCol: prediction column name. (default: prediction)\n","regParam: regularization parameter (>= 0). (default: 0.0, current: 0.3)\n","solver: The solver algorithm for optimization. Supported options: auto, normal, l-bfgs. (default: auto)\n","standardization: whether to standardize the training features before fitting the model. (default: True)\n","tol: the convergence tolerance for iterative algorithms (>= 0). (default: 1e-06)\n","weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0. (undefined)\n"]}]},{"cell_type":"markdown","source":["### Fit a model"],"metadata":{"id":"ihF2_-DbqBDL"}},{"cell_type":"code","source":["linearModel = linear.fit(regDF)"],"metadata":{"id":"5oZ5iR2EgPL4","executionInfo":{"status":"ok","timestamp":1692121917323,"user_tz":-480,"elapsed":1737,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":96,"outputs":[]},{"cell_type":"markdown","source":["### Training Summary\n","[LinearRegressionTrainingSummary](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.LinearRegressionTrainingSummary.html)"],"metadata":{"id":"hCyfbjXnqbGI"}},{"cell_type":"code","source":["linearSummary = linearModel.summary\n","linearSummary"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"rffEaj7tgTrs","executionInfo":{"status":"ok","timestamp":1692121917324,"user_tz":-480,"elapsed":7,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"89152655-9c5b-4676-af53-424657b56569"},"execution_count":97,"outputs":[{"output_type":"execute_result","data":{"text/plain":["<pyspark.ml.regression.LinearRegressionTrainingSummary at 0x7acc400a5fc0>"]},"metadata":{},"execution_count":97}]},{"cell_type":"code","source":["linearSummary.residuals.show()\n","print (linearSummary.totalIterations)\n","print (linearSummary.objectiveHistory)\n","print (linearSummary.rootMeanSquaredError)\n","print (linearSummary.r2)"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"5ysfz2bLqqSS","executionInfo":{"status":"ok","timestamp":1692121917692,"user_tz":-480,"elapsed":372,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"343910ff-05ec-433c-a6bb-98c45d685187"},"execution_count":98,"outputs":[{"output_type":"stream","name":"stdout","text":["+-------------------+\n","|          residuals|\n","+-------------------+\n","|  1.762342343964863|\n","| 0.7623423439648629|\n","|-0.2376576560351371|\n","|-0.2376576560351371|\n","| 0.8547088792080308|\n","+-------------------+\n","\n","5\n","[0.5000000000000001, 0.4315295810362787, 0.3132335933881022, 0.31225692666554117, 0.309150608198303, 0.30915058933480255]\n","0.9518934791114696\n","-0.13262649446867214\n"]}]},{"cell_type":"markdown","source":["##Decision Tree Regressor\n","* [DecisionTreeRegressor](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.DecisionTreeRegressor.html)\n","* [DecisionTreeRegressionModel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.DecisionTreeRegressionModel.html)"],"metadata":{"id":"fWlRbE3NkI47"}},{"cell_type":"code","source":["from pyspark.ml.regression import DecisionTreeRegressor\n","dtr = DecisionTreeRegressor()\n","print (dtr.explainParams())"],"metadata":{"id":"MgfwflVTkL_f","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121917693,"user_tz":-480,"elapsed":9,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"f1461a81-fe17-4d45-9177-45fe4a41b6b7"},"execution_count":99,"outputs":[{"output_type":"stream","name":"stdout","text":["cacheNodeIds: If false, the algorithm will pass trees to executors to match instances with nodes. If true, the algorithm will cache node IDs for each instance. Caching can speed up training of deeper trees. Users can set how often should the cache be checkpointed or disable it by setting checkpointInterval. (default: False)\n","checkpointInterval: set checkpoint interval (>= 1) or disable checkpoint (-1). E.g. 10 means that the cache will get checkpointed every 10 iterations. Note: this setting will be ignored if the checkpoint directory is not set in the SparkContext. (default: 10)\n","featuresCol: features column name. (default: features)\n","impurity: Criterion used for information gain calculation (case-insensitive). Supported options: variance (default: variance)\n","labelCol: label column name. (default: label)\n","leafCol: Leaf indices column name. Predicted leaf index of each instance in each tree by preorder. (default: )\n","maxBins: Max number of bins for discretizing continuous features.  Must be >=2 and >= number of categories for any categorical feature. (default: 32)\n","maxDepth: Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30]. (default: 5)\n","maxMemoryInMB: Maximum memory in MB allocated to histogram aggregation. If too small, then 1 node will be split per iteration, and its aggregates may exceed this size. (default: 256)\n","minInfoGain: Minimum information gain for a split to be considered at a tree node. (default: 0.0)\n","minInstancesPerNode: Minimum number of instances each child must have after split. If a split causes the left or right child to have fewer than minInstancesPerNode, the split will be discarded as invalid. Should be >= 1. (default: 1)\n","minWeightFractionPerNode: Minimum fraction of the weighted sample count that each child must have after split. If a split causes the fraction of the total weight in the left or right child to be less than minWeightFractionPerNode, the split will be discarded as invalid. Should be in interval [0.0, 0.5). (default: 0.0)\n","predictionCol: prediction column name. (default: prediction)\n","seed: random seed. (default: 9073738582406527810)\n","varianceCol: column name for the biased sample variance of prediction. (undefined)\n","weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0. (undefined)\n"]}]},{"cell_type":"code","source":["dtrModel = dtr.fit(regDF)"],"metadata":{"id":"x0Z1T2RrkXc5","executionInfo":{"status":"ok","timestamp":1692121918974,"user_tz":-480,"elapsed":1286,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":100,"outputs":[]},{"cell_type":"markdown","source":["##Random Forest Regressor\n","* [RandomForestRegressor](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.RandomForestRegressor.html)\n","* [RandomForestRegressionModel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.RandomForestRegressionModel.html)"],"metadata":{"id":"ltnUuWndkYGV"}},{"cell_type":"code","source":["from pyspark.ml.regression import RandomForestRegressor\n","rfr =  RandomForestRegressor()\n","print(rfr.explainParams())"],"metadata":{"id":"kCJYI97akg-b","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121918975,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"6ae0bbcc-caf3-4ad3-b18c-fee96d3cc2be"},"execution_count":101,"outputs":[{"output_type":"stream","name":"stdout","text":["bootstrap: Whether bootstrap samples are used when building trees. (default: True)\n","cacheNodeIds: If false, the algorithm will pass trees to executors to match instances with nodes. If true, the algorithm will cache node IDs for each instance. Caching can speed up training of deeper trees. Users can set how often should the cache be checkpointed or disable it by setting checkpointInterval. (default: False)\n","checkpointInterval: set checkpoint interval (>= 1) or disable checkpoint (-1). E.g. 10 means that the cache will get checkpointed every 10 iterations. Note: this setting will be ignored if the checkpoint directory is not set in the SparkContext. (default: 10)\n","featureSubsetStrategy: The number of features to consider for splits at each tree node. Supported options: 'auto' (choose automatically for task: If numTrees == 1, set to 'all'. If numTrees > 1 (forest), set to 'sqrt' for classification and to 'onethird' for regression), 'all' (use all features), 'onethird' (use 1/3 of the features), 'sqrt' (use sqrt(number of features)), 'log2' (use log2(number of features)), 'n' (when n is in the range (0, 1.0], use n * number of features. When n is in the range (1, number of features), use n features). default = 'auto' (default: auto)\n","featuresCol: features column name. (default: features)\n","impurity: Criterion used for information gain calculation (case-insensitive). Supported options: variance (default: variance)\n","labelCol: label column name. (default: label)\n","leafCol: Leaf indices column name. Predicted leaf index of each instance in each tree by preorder. (default: )\n","maxBins: Max number of bins for discretizing continuous features.  Must be >=2 and >= number of categories for any categorical feature. (default: 32)\n","maxDepth: Maximum depth of the tree. (>= 0) E.g., depth 0 means 1 leaf node; depth 1 means 1 internal node + 2 leaf nodes. Must be in range [0, 30]. (default: 5)\n","maxMemoryInMB: Maximum memory in MB allocated to histogram aggregation. If too small, then 1 node will be split per iteration, and its aggregates may exceed this size. (default: 256)\n","minInfoGain: Minimum information gain for a split to be considered at a tree node. (default: 0.0)\n","minInstancesPerNode: Minimum number of instances each child must have after split. If a split causes the left or right child to have fewer than minInstancesPerNode, the split will be discarded as invalid. Should be >= 1. (default: 1)\n","minWeightFractionPerNode: Minimum fraction of the weighted sample count that each child must have after split. If a split causes the fraction of the total weight in the left or right child to be less than minWeightFractionPerNode, the split will be discarded as invalid. Should be in interval [0.0, 0.5). (default: 0.0)\n","numTrees: Number of trees to train (>= 1). (default: 20)\n","predictionCol: prediction column name. (default: prediction)\n","seed: random seed. (default: -5580319131445625397)\n","subsamplingRate: Fraction of the training data used for learning each decision tree, in range (0, 1]. (default: 1.0)\n","weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0. (undefined)\n"]}]},{"cell_type":"code","source":["rfModel = rfr.fit(regDF)"],"metadata":{"id":"5VqOXzaHuLc4","executionInfo":{"status":"ok","timestamp":1692121921426,"user_tz":-480,"elapsed":1924,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":102,"outputs":[]},{"cell_type":"markdown","source":["## Other regression algorithms in MLlib\n","* Generalized Linear Regression:\n","[GeneralizedLinearRegression](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.GeneralizedLinearRegression.html)\n","*  Gradient-Boosted Tree Regressor: [GBTRegressor](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.GBTRegressor.html)\n","* Factorization Machines learning algorithm for regression: [FMRegressor](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.FMRegressor.html)"],"metadata":{"id":"OBr5K9Y_kthv"}},{"cell_type":"markdown","source":["#Clustering\n","* `pyspark.ml.clustering`: [PySpark API: Clustering](https://spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html#clustering)\n","* [Spark Programming Guide: Clustering](https://spark.apache.org/docs/latest/ml-clustering.html)"],"metadata":{"id":"R__eiaIjIwGo"}},{"cell_type":"markdown","source":["## Dataset"],"metadata":{"id":"tPH9GyGOiCOX"}},{"cell_type":"code","source":["retailDF = (spark.read.format(\"csv\")\n","  .option(\"header\", \"true\")\n","  .option(\"inferSchema\", \"true\")\n","  .load(\"/content/SDSC-Spark4/data/online-retail-dataset.csv\")\n","  .where(\"Description IS NOT NULL\"))"],"metadata":{"id":"il8PIyvwATck","executionInfo":{"status":"ok","timestamp":1692121925747,"user_tz":-480,"elapsed":4328,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":103,"outputs":[]},{"cell_type":"code","source":["retailDF.printSchema()"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"DJ4LawUsAfc_","executionInfo":{"status":"ok","timestamp":1692121925747,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"5fc385fe-2def-4bf8-e47b-9c18ce54f135"},"execution_count":104,"outputs":[{"output_type":"stream","name":"stdout","text":["root\n"," |-- InvoiceNo: string (nullable = true)\n"," |-- StockCode: string (nullable = true)\n"," |-- Description: string (nullable = true)\n"," |-- Quantity: integer (nullable = true)\n"," |-- InvoiceDate: string (nullable = true)\n"," |-- UnitPrice: double (nullable = true)\n"," |-- CustomerID: integer (nullable = true)\n"," |-- Country: string (nullable = true)\n","\n"]}]},{"cell_type":"code","source":["from pyspark.ml.feature import VectorAssembler\n","\n","va = VectorAssembler()\\\n","  .setInputCols([\"Quantity\", \"UnitPrice\"])\\\n","  .setOutputCol(\"features\")\n","\n","retail_clusteringDF = va.transform(retailDF).select(\"features\")"],"metadata":{"id":"zFnA0x8_I_5a","executionInfo":{"status":"ok","timestamp":1692121925748,"user_tz":-480,"elapsed":6,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":105,"outputs":[]},{"cell_type":"code","source":["retail_clusteringDF.show(2)"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"THpj_Rer0pVv","executionInfo":{"status":"ok","timestamp":1692121926297,"user_tz":-480,"elapsed":555,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"885ee6c9-2ed6-42b8-eb6b-f03e1530eab4"},"execution_count":106,"outputs":[{"output_type":"stream","name":"stdout","text":["+----------+\n","|  features|\n","+----------+\n","|[6.0,2.55]|\n","|[6.0,3.39]|\n","+----------+\n","only showing top 2 rows\n","\n"]}]},{"cell_type":"markdown","source":["##K-means\n","\n","* [KMeans](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.clustering.KMeans.html) is implemented based on [Scalable K-Means++](http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf), known as (***k-means||***).\n","* [KMeansModel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.clustering.KMeansModel.html): a model fitted by KMeans.\n","* [KMeansSummary](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.clustering.KMeansSummary.html): a summary of KMeans."],"metadata":{"id":"9GM5OurQI1gs"}},{"cell_type":"code","source":["from pyspark.ml.clustering import KMeans\n","km = KMeans().setK(5)"],"metadata":{"id":"hwNSPiieJTXB","executionInfo":{"status":"ok","timestamp":1692121926298,"user_tz":-480,"elapsed":9,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":107,"outputs":[]},{"cell_type":"markdown","source":["### Parameters\n"],"metadata":{"id":"g7u-HdrJsJCG"}},{"cell_type":"code","source":["print(km.explainParams())"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"bTbpLfi60AKm","executionInfo":{"status":"ok","timestamp":1692121926299,"user_tz":-480,"elapsed":10,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"2ff16445-8d35-4ae8-f60c-1aa80f0db6b7"},"execution_count":108,"outputs":[{"output_type":"stream","name":"stdout","text":["distanceMeasure: the distance measure. Supported options: 'euclidean' and 'cosine'. (default: euclidean)\n","featuresCol: features column name. (default: features)\n","initMode: The initialization algorithm. This can be either \"random\" to choose random points as initial cluster centers, or \"k-means||\" to use a parallel variant of k-means++ (default: k-means||)\n","initSteps: The number of steps for k-means|| initialization mode. Must be > 0. (default: 2)\n","k: The number of clusters to create. Must be > 1. (default: 2, current: 5)\n","maxBlockSizeInMB: maximum memory in MB for stacking input data into blocks. Data is stacked within partitions. If more than remaining data size in a partition then it is adjusted to the data size. Default 0.0 represents choosing optimal value, depends on specific algorithm. Must be >= 0. (default: 0.0)\n","maxIter: max number of iterations (>= 0). (default: 20)\n","predictionCol: prediction column name. (default: prediction)\n","seed: random seed. (default: -7685785370690492299)\n","solver: The solver algorithm for optimization. Supported options: auto, row, block. (default: auto)\n","tol: the convergence tolerance for iterative algorithms (>= 0). (default: 0.0001)\n","weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0. (undefined)\n"]}]},{"cell_type":"markdown","source":["### Fit a model"],"metadata":{"id":"juqEyQ-D0Svm"}},{"cell_type":"code","source":["kmModel = km.fit(retail_clusteringDF)"],"metadata":{"id":"t-i4ffqPJV3H","executionInfo":{"status":"ok","timestamp":1692121949965,"user_tz":-480,"elapsed":23672,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}}},"execution_count":109,"outputs":[]},{"cell_type":"code","source":["print(\"Cluster Centers: \")\n","for center in kmModel.clusterCenters():\n","    print(center)"],"metadata":{"colab":{"base_uri":"https://localhost:8080/"},"id":"t4e6zPWJDNsR","executionInfo":{"status":"ok","timestamp":1692121949966,"user_tz":-480,"elapsed":15,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"8787b281-55da-4ee1-aec1-c0c51dfc284a"},"execution_count":110,"outputs":[{"output_type":"stream","name":"stdout","text":["Cluster Centers: \n","[9.0114221  4.55313107]\n","[-7.7605e+04  1.5600e+00]\n","[7.7605e+04 1.5600e+00]\n","[1.19799628e+03 1.16297398e+00]\n","[-1.000e+00  3.897e+04]\n"]}]},{"cell_type":"markdown","source":["### Training Summary"],"metadata":{"id":"n94fX0Py1RFl"}},{"cell_type":"code","source":["summary = kmModel.summary\n","\n","print(\"Cluster Sizes: \")\n","print (summary.clusterSizes) # number of points in each cluster"],"metadata":{"id":"IKnPPL3gJklg","colab":{"base_uri":"https://localhost:8080/"},"executionInfo":{"status":"ok","timestamp":1692121949967,"user_tz":-480,"elapsed":14,"user":{"displayName":"Salman Salloum","userId":"10391530522294962645"}},"outputId":"19f53933-1b8a-486d-e8de-1660d5484e30"},"execution_count":111,"outputs":[{"output_type":"stream","name":"stdout","text":["Cluster Sizes: \n","[540181, 2, 2, 269, 1]\n"]}]},{"cell_type":"markdown","source":["## Other clustering algorithms in MLlib\n","* Bisecting K-Means: [BisectingKMeans](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.mllib.clustering.BisectingKMeans.html)\n","\n","* Gaussian Texture Models: [GaussianMixture](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.clustering.GaussianMixture.html)\n","\n","* Latent Dirichlet allocation: [LDA](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.clustering.LDA.html)"],"metadata":{"id":"qEkCRKnDJrLo"}}],"metadata":{"application/vnd.databricks.v1+notebook":{"dashboards":[],"language":"python","notebookMetadata":{"mostRecentlyExecutedCommandWithImplicitDF":{"commandId":3496694700218674,"dataframes":["_sqldf"]},"pythonIndentUnit":4},"notebookName":"1-SparkSQL_DataFrames","widgets":{}},"colab":{"provenance":[{"file_id":"1-3DVy-jpDMwUjlHuvXT5nkt0fWHazHhZ","timestamp":1691337251552}],"toc_visible":true},"kernelspec":{"display_name":"Python 3","name":"python3"},"language_info":{"name":"python"}},"nbformat":4,"nbformat_minor":0}