ekSpark.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "ekSpark.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "authorship_tag": "ABX9TyMBCEBa6J9AXPwycc/3vFFg",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/kgene521/Calc/blob/master/ekSpark.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ZgBtBdgCF8P5"
      },
      "outputs": [],
      "source": [
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n",
        "!wget -q http://archive.apache.org/dist/spark/spark-3.2.1/spark-3.2.1-bin-hadoop3.2.tgz\n",
        "!wget https://jacobceles.github.io/knowledge_repo/colab_and_pyspark/cars.csv\n",
        "\n",
        "!mkdir spark321\n",
        "!tar xf spark-3.2.1-bin-hadoop3.2.tgz -C spark321/\n",
        "!wget https://jacobceles.github.io/knowledge_repo/colab_and_pyspark/cars.csv\n",
        "!mv cars.csv spark321/\n",
        "!wget -q https://github.com/apache/spark/blob/master/examples/src/main/resources/dir1/file1.parquet\n",
        "!wget -q https://github.com/apache/spark/blob/master/examples/src/main/resources/dir1/dir2/file2.parquet\n",
        "\n",
        "!mv file*.parquet spark321/\n",
        "\n",
        "!pip install -q findspark\n",
        "!pip install pyspark\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark321/spark-3.2.1-bin-hadoop3.2\"\n",
        "!echo $SPARK_HOME\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!pwd"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "p5QfE_pu3S6p",
        "outputId": "65c3b2bc-2840-4242-f2b3-3431505c591f"
      },
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "/content\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import os\n",
        "import findspark\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.types import *\n",
        "from pyspark.sql.functions import col, lit, concat, lower, upper, substring\n",
        "from pyspark.sql.functions import min, max, to_date, to_timestamp, lit, date_add, date_sub\n",
        "from pyspark.sql import functions\n",
        "from datetime import datetime\n",
        "\n",
        "findspark.init()\n",
        "spark = SparkSession.builder.master(\"local[*]\").getOrCreate()\n",
        "spark.conf.set(\"spark.sql.repl.eagerEval.enabled\", 'True') # Property used to format output tables better\n",
        "# spark\n",
        "def print_functions():\n",
        "    foos = dir(functions)\n",
        "    function_list = \"\"\n",
        "    counter = 0\n",
        "    for x in foos:\n",
        "      counter = counter + 1\n",
        "      if x[0] == '_' or counter <= 8:\n",
        "        continue\n",
        "      if counter % 10 == 0:\n",
        "        print(function_list)\n",
        "        function_list = \"\"\n",
        "      else:\n",
        "        function_list = function_list + \", \" + x\n",
        "    print(function_list)\n",
        "\n",
        "def cars():\n",
        "  df = spark.read.csv('/content/spark321/cars.csv', header=True, sep=\";\")\n",
        "    # df.show(5, truncate=False)\n",
        "  df2 = df.limit(5)\n",
        "  # df2\n",
        "  # df2.columns\n",
        "  # df2.dtypes\n",
        "  # df2.printSchema\n",
        "  df.limit(5)\n",
        "  # Creating a list of the schema in the format column_name, data_type\n",
        "  labels = [\n",
        "      ('Car',StringType()),\n",
        "      ('MPG',DoubleType()),\n",
        "      ('Cylinders',IntegerType()),\n",
        "      ('Displacement',DoubleType()),\n",
        "      ('Horsepower',DoubleType()),\n",
        "      ('Weight',DoubleType()),\n",
        "      ('Acceleration',DoubleType()),\n",
        "      ('Model',IntegerType()),\n",
        "      ('Origin',StringType())\n",
        "  ]\n",
        "  # # Creating the schema that will be passed when reading the csv\n",
        "  # print(labels)\n",
        "  schema = StructType([StructField (x[0], x[1], True) for x in labels])\n",
        "  # schema\n",
        "  df = spark.read.csv('/content/spark321/cars.csv', header=True, sep=\";\", schema=schema)\n",
        "  # df.printSchema()\n",
        "  # df.show(truncate=False)\n",
        "  # print(df.Car)\n",
        "  # print(\"*\"*35)\n",
        "  # df.select(df.Car).show(truncate=False)\n",
        "  # print(df['car'])\n",
        "  # print(\"*\"*20)\n",
        "  # Column name is case sensitive\n",
        "  # df.select(df['car']).show(truncate=False)\n",
        "\n",
        "  ### Column name is case insensitive\n",
        "  # df.select(col('car'), col('CAR')).show(truncate=False)\n",
        "  ### 1st method\n",
        "  ### Column name is case sensitive in this usage\n",
        "  ### print(df.Car, df.Cylinders)\n",
        "  ### print(\"*\"*40)\n",
        "  # df.select(df.Car, df.Cylinders).show(truncate=False)\n",
        "  ### print(df2)\n",
        "  ### 2nd method\n",
        "  ### Column name is case insensitive in this usage\n",
        "  ### print(df['car'],df['cylinders'])\n",
        "  ### print(\"*\"*40)\n",
        "  ### df.select(df['caR'],df['cylINDers']).show(truncate=False)\n",
        "  ### 3rd method\n",
        "  ### Column name is case insensitive in this usage\n",
        "  \n",
        "  # df.select(col('caR'),col('cylINDers')).show(truncate=False)\n",
        "  ### CASE 1: Adding a new column\n",
        "  ### We will add a new column called 'first_column' at the end\n",
        "  \n",
        "  df = df.withColumn('first_column',lit(1))\n",
        "  ### lit means literal. It populates the row with the literal value given.\n",
        "  ### When adding static data / constant values, it is a good practice to use it.\n",
        "  # df.show(5,truncate=False)\n",
        "  # CASE 2: Adding multiple columns\n",
        "  # We will add two new columns called 'second_column' and 'third_column' at the end\n",
        "  df = df.withColumn('second_column', lit(2)) \\\n",
        "        .withColumn('third_column', lit('Third Column'))\n",
        "  # lit means literal. It populates the row with the literal value given.\n",
        "  # When adding static data / constant values, it is a good practice to use it.\n",
        "  # df.show(5,truncate=False)\n",
        "  # CASE 3: Deriving a new column from an exisitng one\n",
        "  # We will add a new column called 'car_model' which has the value of car and model appended together with a space in between \n",
        "  \n",
        "  df = df.withColumn('car_model', concat(col(\"Car\"), lit(\" \"), col(\"model\")))\n",
        "  # lit means literal. It populates the row with the literal value given.\n",
        "  # When adding static data / constant values, it is a good practice to use it.\n",
        "  # df.show(5,truncate=False)\n",
        "  #Renaming a column in PySpark\n",
        "  df = df.withColumnRenamed('first_column', 'new_column_one') \\\n",
        "        .withColumnRenamed('second_column', 'new_column_two') \\\n",
        "        .withColumnRenamed('third_column', 'new_column_three')\n",
        "  # df.show(truncate=False)\n",
        "\n",
        "  # Group By a column in PySpark\n",
        "  # df.groupBy('Origin').count().show(5)\n",
        "\n",
        "  # Group By multiple columns in PySpark\n",
        "  # df.groupBy('Origin', 'Model').count().show(5)\n",
        "  #Remove columns in PySpark\n",
        "  df = df.drop('new_column_one')\n",
        "  # df.show(5,truncate=False)\n",
        "\n",
        "  #Remove multiple columnss in one go\n",
        "  df = df.drop('new_column_two') \\\n",
        "        .drop('new_column_three')\n",
        "  # df.show(5,truncate=False)\n",
        "\n",
        "\n",
        "  # Filtering rows in PySpark\n",
        "  total_count = df.count()\n",
        "  # print(\"TOTAL RECORD COUNT: \" + str(total_count))\n",
        "  europe_filtered_count = df.filter(col('Origin')=='Europe').count()\n",
        "  # print(\"EUROPE FILTERED RECORD COUNT: \" + str(europe_filtered_count))\n",
        "  # df.filter(col('Origin')=='Europe').show(truncate=False)\n",
        "  # Filtering rows in PySpark based on Multiple conditions\n",
        "  total_count = df.count()\n",
        "  # print(\"TOTAL RECORD COUNT: \" + str(total_count))\n",
        "  europe_filtered_count = df.filter((col('Origin')=='Europe') &\n",
        "                                    (col('Cylinders')==4)).count() # Two conditions added here\n",
        "  # print(\"EUROPE FILTERED RECORD COUNT: \" + str(europe_filtered_count))\n",
        "  # df.filter(col('Origin')=='Europe').show(truncate=False)\n",
        "\n",
        "  #Get Unique Rows in PySpark\n",
        "  # df.select('Origin').distinct().show()\n",
        "\n",
        "\n",
        "\n",
        "  #Get Unique Rows in PySpark based on mutliple columns\n",
        "  # df.select('Origin','model').distinct().show()\n",
        "\n",
        "\n",
        "\n",
        "  # Sort Rows in PySpark\n",
        "  # By default the data will be sorted in ascending order\n",
        "  # df.orderBy('Cylinders').show(5, truncate=False)\n",
        "  # df.orderBy('Horsepower', ascending=False).show(5, truncate=False)\n",
        "\n",
        "  # Using groupBy aand orderBy together\n",
        "  # df.groupBy(\"Origin\").count().orderBy('count', ascending=False).show(10)\n",
        "\n",
        "  # CASE 1: Union When columns are in order\n",
        "  europe_cars = df.filter((col('Origin')=='Europe') & (col('Cylinders')==5))\n",
        "  japan_cars = df.filter((col('Origin')=='Japan') & (col('Cylinders')==3))\n",
        "  # print(\"EUROPE CARS: \"+str(europe_cars.count()))\n",
        "  # print(\"JAPAN CARS: \"+str(japan_cars.count()))\n",
        "  # print(\"AFTER UNION: \"+str(europe_cars.union(japan_cars).count()))\n",
        "\n",
        "  # CASE 1: Union When columns are not in order\n",
        "  # Creating two dataframes with jumbled columns\n",
        "  df1 = spark.createDataFrame([[1, 2, 3]], [\"col0\", \"col1\", \"col2\"])\n",
        "  df2 = spark.createDataFrame([[4, 5, 6]], [\"col1\", \"col2\", \"col0\"])\n",
        "  # df1.unionByName(df2).show()\n",
        "\n",
        "  # Functions available in PySpark ######################### Functions in Spark ###\n",
        "  \n",
        "  # Similar to python, we can use the dir function to view the avaiable functions\n",
        "    # print(foos)\n",
        "  # print_functions()\n",
        "\n",
        "  # Loading the data\n",
        "  df = spark.read.csv('/content/spark321/cars.csv', header=True, sep=\";\", inferSchema=True)\n",
        "\n",
        "    ### Prints out the details of a function\n",
        "  ###       help(substring)  ### HELP on the substring() function\n",
        "  ### alias is used to rename the column name in the output\n",
        "  # df.select(col('Car'),\n",
        "  #           lower(col('Car')),\n",
        "  #           upper(col('Car')),\n",
        "  #           substring(col('Car'),1,4)\n",
        "  #           .alias(\"concatenated value\")\n",
        "  #           ).show(5, False)\n",
        "\n",
        "  # df.select(col(\"Car\"),\n",
        "  #           col(\"model\"),\n",
        "  #           concat(col(\"Car\"), lit(\" \"), col(\"model\"))\n",
        "  #           ).show(5, False)\n",
        "\n",
        "  # df.select(min(col('Weight')), max(col('Weight'))).show()\n",
        "\n",
        "  # df.select(min(col('Weight'))+lit(10), max(col('Weight')+lit(10))).show()\n",
        "\n",
        "  df = spark.createDataFrame([('2019-12-25 13:30:00',)], ['DOB'])\n",
        "  # df.show()\n",
        "  # df.printSchema()\n",
        "\n",
        "  ###########################  '2019-12-25 13:30:33',\n",
        "  df = spark.createDataFrame([(datetime.now()       ,)], ['DOB'])\n",
        "  df = df.select(to_date(col('DOB'),'yyyy-MM-dd HH:mm:ss'), to_timestamp(col('DOB'),'yyyy-MM-dd HH:mm:ss'))\n",
        "  # df.show()\n",
        "  # df.printSchema()\n",
        "\n",
        "  df = spark.createDataFrame([('25/Dec/2019 13:30:00',)], ['DOB'])\n",
        "  df = df.select(to_date(col('DOB'),'dd/MMM/yyyy HH:mm:ss'), to_timestamp(col('DOB'),'dd/MMM/yyyy HH:mm:ss'))\n",
        "  # df.show()\n",
        "  # df.printSchema()\n",
        "\n",
        "  ### What is 3 days earlier that the oldest date and 3 days later than the most recent date?\n",
        "\n",
        "  # create a dummy dataframe\n",
        "  df = spark.createDataFrame([('1990-01-01',),('1995-01-03',),('2021-03-30',)], ['Date'])\n",
        "  # find out the required dates\n",
        "  # df.select(date_add(max(col('Date')),3),\n",
        "  #           date_sub(min(col('Date')),3)\n",
        "  #           ).show()\n",
        "\n",
        "  ### Joins in PySpark\n",
        "\n",
        "  print('*'*10, \"Create two dataframes \", '*'*10)\n",
        "  cars_df = spark.createDataFrame([[1, 'Car A'],[2, 'Car B'],[3, 'Car C'], [4, 'Car D']], [\"id\", \"car_name\"])\n",
        "  car_price_df = spark.createDataFrame([[1, 1000],[2, 2000],[3, 3000]], [\"id\", \"car_price\"])\n",
        "  # cars_df.show()\n",
        "  # car_price_df.show()\n",
        "\n",
        "  # Executing an inner join so we can see the id, name and price of each car in one row\n",
        "  cdf = cars_df.join(car_price_df,\n",
        "                    cars_df.id == car_price_df.id,\n",
        "                    'inner'  ### left, right, outer, full, \n",
        "                    ).select(cars_df['id'],\n",
        "                            cars_df['car_name'],\n",
        "                            car_price_df['car_price'])\n",
        "                            # .show(truncate=False)\n",
        "\n",
        "  # cdf.orderBy(cdf.cars_df['car_price'], ascending=False)\n",
        "  # cdf.show(truncate=False)\n",
        "\n",
        "# cars()\n",
        "# print_functions()\n",
        "\n",
        "people = [\n",
        "  {\"name\":\"Michael\", \"salary\":3000},\n",
        "  {\"name\":\"Andy\", \"salary\":4500},\n",
        "  {\"name\":\"Justin\", \"salary\":3500},\n",
        "  {\"name\":\"Berta\", \"salary\":4000}\n",
        "]\n",
        "\n",
        "# print(people)"
      ],
      "metadata": {
        "id": "uq34zSAEc83h"
      },
      "execution_count": 16,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql import SparkSession\n",
        "\n",
        "# spark = SparkSession.builder.appName(\"pySpark Training\").config(\"key1\", \"10\").getOrCreate()\n",
        "\n",
        "# df = spark.read.csv(\"/content/spark321/cars.csv\", header=True, sep=\";\")\n",
        "### Create Temp Table\n",
        "# df.createOrReplaceTempView(\"temp\")\n",
        "### SELECT * FROM TEMP\n",
        "# spark.sql(\"select * from temp limit 5\").show()\n",
        "### SELECT COUNT(*) FROM TEMP\n",
        "# spark.sql(\"select count(*) as total_count from temp\").show()\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n"
      ],
      "metadata": {
        "id": "1bIASJN9KgUC",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "collapsed": true,
        "outputId": "688eb2f7-362f-45e8-8062-13b83db81019"
      },
      "execution_count": 20,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
            "|                 Car| MPG|Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|\n",
            "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
            "|Chevrolet Chevell...|18.0|        8|       307.0|     130.0| 3504.|        12.0|   70|    US|\n",
            "|   Buick Skylark 320|15.0|        8|       350.0|     165.0| 3693.|        11.5|   70|    US|\n",
            "|  Plymouth Satellite|18.0|        8|       318.0|     150.0| 3436.|        11.0|   70|    US|\n",
            "|       AMC Rebel SST|16.0|        8|       304.0|     150.0| 3433.|        12.0|   70|    US|\n",
            "|         Ford Torino|17.0|        8|       302.0|     140.0| 3449.|        10.5|   70|    US|\n",
            "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
            "\n",
            "+-----------+\n",
            "|total_count|\n",
            "+-----------+\n",
            "|        406|\n",
            "+-----------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "pips = [['Bob', 85],['Alice',None], [None, 80] ]\n",
        "xnames = spark.createDataFrame(pips).collect()\n",
        "xnames\n",
        "xnames = xnames.withColumnRenamed('_1', 'Name')"
      ],
      "metadata": {
        "id": "OTsQ-3b8B1nO",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "81884547-3abb-4cf3-b6a9-1e943a605c10"
      },
      "execution_count": 6,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[Row(_1='Bob', _2=85), Row(_1='Alice', _2=None), Row(_1=None, _2=80)]"
            ]
          },
          "metadata": {},
          "execution_count": 6
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!hdfs dfs -ls"
      ],
      "metadata": {
        "id": "M6NqFrJ7ab10",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "3254f004-1f63-4d74-a4e4-1c56a4171fdd"
      },
      "execution_count": 21,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "/bin/bash: hdfs: command not found\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        ""
      ],
      "metadata": {
        "id": "ncSMIU4bZ2l8"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}