Sparkpractice(121222).ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "d3d22efe",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "from pyspark.sql import SparkSession\n",
    "spark=SparkSession.builder.master('local[*]').appName('First_program').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "896f6bce",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: struct (nullable = true)\n",
      " |    |-- firstname: string (nullable = true)\n",
      " |    |-- middlename: string (nullable = true)\n",
      " |    |-- lastname: string (nullable = true)\n",
      " |-- languages: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      "\n",
      "+----------------------+------------------+-----+------+\n",
      "|name                  |languages         |state|gender|\n",
      "+----------------------+------------------+-----+------+\n",
      "|{James, , Smith}      |[Java, Scala, C++]|OH   |M     |\n",
      "|{Anna, Rose, }        |[Spark, Java, C++]|NY   |F     |\n",
      "|{Julia, , Williams}   |[CSharp, VB]      |OH   |F     |\n",
      "|{Maria, Anne, Jones}  |[CSharp, VB]      |NY   |M     |\n",
      "|{Jen, Mary, Brown}    |[CSharp, VB]      |NY   |M     |\n",
      "|{Mike, Mary, Williams}|[Python, VB]      |OH   |M     |\n",
      "+----------------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import StructType,StructField \n",
    "from pyspark.sql.types import StringType, IntegerType, ArrayType\n",
    "data = [\n",
    "    ((\"James\",\"\",\"Smith\"),[\"Java\",\"Scala\",\"C++\"],\"OH\",\"M\"),\n",
    "    ((\"Anna\",\"Rose\",\"\"),[\"Spark\",\"Java\",\"C++\"],\"NY\",\"F\"),\n",
    "    ((\"Julia\",\"\",\"Williams\"),[\"CSharp\",\"VB\"],\"OH\",\"F\"),\n",
    "    ((\"Maria\",\"Anne\",\"Jones\"),[\"CSharp\",\"VB\"],\"NY\",\"M\"),\n",
    "    ((\"Jen\",\"Mary\",\"Brown\"),[\"CSharp\",\"VB\"],\"NY\",\"M\"),\n",
    "    ((\"Mike\",\"Mary\",\"Williams\"),[\"Python\",\"VB\"],\"OH\",\"M\")\n",
    " ]\n",
    "        \n",
    "schema = StructType([\n",
    "     StructField('name', StructType([\n",
    "        StructField('firstname', StringType(), True),\n",
    "        StructField('middlename', StringType(), True),\n",
    "         StructField('lastname', StringType(), True)\n",
    "     ])),\n",
    "     StructField('languages', ArrayType(StringType()), True),\n",
    "     StructField('state', StringType(), True),\n",
    "     StructField('gender',StringType())\n",
    " ])\n",
    "\n",
    "df = spark.createDataFrame(data = data, schema = schema)\n",
    "df.printSchema()\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "949b42a6",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- state: string (nullable = true)\n",
      "\n",
      "+--------------------+------------------+-----+------+\n",
      "|                name|         languages|state|gender|\n",
      "+--------------------+------------------+-----+------+\n",
      "|    {James, , Smith}|[Java, Scala, C++]|   OH|     M|\n",
      "|{Mike, Mary, Will...|      [Python, VB]|   OH|     M|\n",
      "+--------------------+------------------+-----+------+\n",
      "\n",
      "+--------------------+------------------+-----+------+\n",
      "|                name|         languages|state|gender|\n",
      "+--------------------+------------------+-----+------+\n",
      "|    {James, , Smith}|[Java, Scala, C++]|   OH|     M|\n",
      "|      {Anna, Rose, }|[Spark, Java, C++]|   NY|     F|\n",
      "| {Julia, , Williams}|      [CSharp, VB]|   OH|     F|\n",
      "|{Maria, Anne, Jones}|      [CSharp, VB]|   NY|     M|\n",
      "|  {Jen, Mary, Brown}|      [CSharp, VB]|   NY|     M|\n",
      "|{Mike, Mary, Will...|      [Python, VB]|   OH|     M|\n",
      "+--------------------+------------------+-----+------+\n",
      "\n",
      "+--------------------+------------------+-----+------+\n",
      "|                name|         languages|state|gender|\n",
      "+--------------------+------------------+-----+------+\n",
      "|    {James, , Smith}|[Java, Scala, C++]|   OH|     M|\n",
      "| {Julia, , Williams}|      [CSharp, VB]|   OH|     F|\n",
      "|{Mike, Mary, Will...|      [Python, VB]|   OH|     M|\n",
      "+--------------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df.state).printSchema()\n",
    "df.filter((df.state=='OH') & (df.gender=='M')).show()\n",
    "df.filter(df.state.isin(['OH','NY'])).show()\n",
    "df.filter(df.state.like('O%')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "68b95acb",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+------+\n",
      "| name|state|gender|\n",
      "+-----+-----+------+\n",
      "|James| null|     M|\n",
      "| Anna|   NY|     F|\n",
      "|Julia| null|  null|\n",
      "+-----+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data = [\n",
    "    (\"James\",None,\"M\"),\n",
    "    (\"Anna\",\"NY\",\"F\"),\n",
    "    (\"Julia\",None,None)\n",
    "  ]\n",
    "\n",
    "columns = [\"name\",\"state\",\"gender\"]\n",
    "df1 = spark.createDataFrame(data,columns)\n",
    "df1.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "c72efabf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+------+\n",
      "| name|state|gender|\n",
      "+-----+-----+------+\n",
      "|Julia| null|  null|\n",
      "+-----+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1.filter(df1.state.isNull() & df1.gender.isNull()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "58adc752",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+------+\n",
      "|name|state|gender|\n",
      "+----+-----+------+\n",
      "|Anna|   NY|     F|\n",
      "+----+-----+------+\n",
      "\n",
      "+-----+-----+------+\n",
      "| name|state|gender|\n",
      "+-----+-----+------+\n",
      "|James| null|     M|\n",
      "| Anna|   NY|     F|\n",
      "+-----+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1.na.drop().show()\n",
    "df1.na.drop(subset=['gender']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "70719d0e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- firstname: string (nullable = true)\n",
      " |-- middlename: string (nullable = true)\n",
      " |-- lastname: string (nullable = true)\n",
      " |-- dob: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- salary: string (nullable = true)\n",
      "\n",
      "+---------+----------+--------+----------+------+------+\n",
      "|firstname|middlename|lastname|       dob|gender|salary|\n",
      "+---------+----------+--------+----------+------+------+\n",
      "|    James|       xxx|   Smith|1991-04-01|     M|  3000|\n",
      "|  Michael|      Rose|     yyy|2000-05-19|     M|  4000|\n",
      "|   Robert|       aaa|Williams|1978-09-05|     M|  4000|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|  4000|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|    -1|\n",
      "+---------+----------+--------+----------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data = [('James','xxx','Smith','1991-04-01','M','3000'),\n",
    "  ('Michael','Rose','yyy','2000-05-19','M','4000'),\n",
    "  ('Robert','aaa','Williams','1978-09-05','M','4000'),\n",
    "  ('Maria','Anne','Jones','1967-12-01','F','4000'),\n",
    "  ('Jen','Mary','Brown','1980-02-17','F','-1')]\n",
    "\n",
    "columns = [\"firstname\",\"middlename\",\"lastname\",\"dob\",\"gender\",\"salary\"]\n",
    "\n",
    "df2 = spark.createDataFrame(data=data, schema=columns)\n",
    "\n",
    "df2.printSchema()\n",
    "df2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "0abcd11b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- dob: date (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_3=df2.withColumn('dob',df2.dob.cast('date'))\n",
    "df_3.select(df_3.dob).printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "d7f412ca",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+------+------+------------+\n",
      "|firstname|middlename|lastname|       dob|gender|salary|Updatesalary|\n",
      "+---------+----------+--------+----------+------+------+------------+\n",
      "|    James|       xxx|   Smith|1991-04-01|     M|  3000|      4000.0|\n",
      "|  Michael|      Rose|     yyy|2000-05-19|     M|  4000|      5000.0|\n",
      "|   Robert|       aaa|Williams|1978-09-05|     M|  4000|      5000.0|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|  4000|      5000.0|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|    -1|       999.0|\n",
      "+---------+----------+--------+----------+------+------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.withColumn('Updatesalary',col('salary')+1000).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "003da0bf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+------+------+---------------+\n",
      "|firstname|middlename|lastname|       dob|gender|salary|           Name|\n",
      "+---------+----------+--------+----------+------+------+---------------+\n",
      "|    James|       xxx|   Smith|1991-04-01|     M|  3000|    James-Smith|\n",
      "|  Michael|      Rose|     yyy|2000-05-19|     M|  4000|    Michael-yyy|\n",
      "|   Robert|       aaa|Williams|1978-09-05|     M|  4000|Robert-Williams|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|  4000|    Maria-Jones|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|    -1|      Jen-Brown|\n",
      "+---------+----------+--------+----------+------+------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import concat,lit\n",
    "df2.withColumn('Name',concat(col('firstname'),lit('-'),col('lastname'))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "8a0c2cb8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+-----------+------+------+\n",
      "|firstname|middlename|lastname|DateOfBirth|gender|salary|\n",
      "+---------+----------+--------+-----------+------+------+\n",
      "|    James|       xxx|   Smith| 1991-04-01|     M|  3000|\n",
      "|  Michael|      Rose|     yyy| 2000-05-19|     M|  4000|\n",
      "|   Robert|       aaa|Williams| 1978-09-05|     M|  4000|\n",
      "|    Maria|      Anne|   Jones| 1967-12-01|     F|  4000|\n",
      "|      Jen|      Mary|   Brown| 1980-02-17|     F|    -1|\n",
      "+---------+----------+--------+-----------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.withColumnRenamed('dob','DateOfBirth').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "e652fa0c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- employee_name: string (nullable = true)\n",
      " |-- department: string (nullable = true)\n",
      " |-- salary: long (nullable = true)\n",
      "\n",
      "+-------------+----------+------+\n",
      "|employee_name|department|salary|\n",
      "+-------------+----------+------+\n",
      "|James        |Sales     |3000  |\n",
      "|Michael      |Sales     |4600  |\n",
      "|Robert       |Sales     |4100  |\n",
      "|Maria        |Finance   |3000  |\n",
      "|James        |Sales     |3000  |\n",
      "|Scott        |Finance   |3300  |\n",
      "|Jen          |Finance   |3900  |\n",
      "|Jeff         |Marketing |3000  |\n",
      "|Kumar        |Marketing |2000  |\n",
      "|Saif         |Sales     |4100  |\n",
      "+-------------+----------+------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data = [(\"James\", \"Sales\", 3000), \\\n",
    "    (\"Michael\", \"Sales\", 4600), \\\n",
    "    (\"Robert\", \"Sales\", 4100), \\\n",
    "    (\"Maria\", \"Finance\", 3000), \\\n",
    "    (\"James\", \"Sales\", 3000), \\\n",
    "    (\"Scott\", \"Finance\", 3300), \\\n",
    "    (\"Jen\", \"Finance\", 3900), \\\n",
    "    (\"Jeff\", \"Marketing\", 3000), \\\n",
    "    (\"Kumar\", \"Marketing\", 2000), \\\n",
    "    (\"Saif\", \"Sales\", 4100) \\\n",
    "  ]\n",
    "columns= [\"employee_name\", \"department\", \"salary\"]\n",
    "df3 = spark.createDataFrame(data = data, schema = columns)\n",
    "df3.printSchema()\n",
    "df3.show(truncate=False)\n",
    "df3.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "e4464763",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|department|\n",
      "+----------+\n",
      "|     Sales|\n",
      "|   Finance|\n",
      "| Marketing|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df3.select(df3.department).distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "504544bc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_4=df3.select(df3.salary).dropDuplicates()\n",
    "df_4.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "0621e853",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "9"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df3.select(df3.department,df3.employee_name).distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "540f948e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "9"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df3.select(df3.department,df3.employee_name).dropDuplicates().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "85b6e71b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+----------+------+\n",
      "|employee_name|department|salary|\n",
      "+-------------+----------+------+\n",
      "|      Michael|     Sales|  4600|\n",
      "|         Saif|     Sales|  4100|\n",
      "|       Robert|     Sales|  4100|\n",
      "|          Jen|   Finance|  3900|\n",
      "|        Scott|   Finance|  3300|\n",
      "|        James|     Sales|  3000|\n",
      "|        Maria|   Finance|  3000|\n",
      "|         Jeff| Marketing|  3000|\n",
      "|        James|     Sales|  3000|\n",
      "|        Kumar| Marketing|  2000|\n",
      "+-------------+----------+------+\n",
      "\n",
      "+-------------+----------+------+\n",
      "|employee_name|department|salary|\n",
      "+-------------+----------+------+\n",
      "|      Michael|     Sales|  4600|\n",
      "|         Saif|     Sales|  4100|\n",
      "|       Robert|     Sales|  4100|\n",
      "|          Jen|   Finance|  3900|\n",
      "|        Scott|   Finance|  3300|\n",
      "|        Maria|   Finance|  3000|\n",
      "|         Jeff| Marketing|  3000|\n",
      "|        James|     Sales|  3000|\n",
      "|        James|     Sales|  3000|\n",
      "|        Kumar| Marketing|  2000|\n",
      "+-------------+----------+------+\n",
      "\n",
      "+-------------+----------+------+\n",
      "|employee_name|department|salary|\n",
      "+-------------+----------+------+\n",
      "|        Kumar| Marketing|  2000|\n",
      "|        Maria|   Finance|  3000|\n",
      "|        James|     Sales|  3000|\n",
      "|        James|     Sales|  3000|\n",
      "|         Jeff| Marketing|  3000|\n",
      "|        Scott|   Finance|  3300|\n",
      "|          Jen|   Finance|  3900|\n",
      "|       Robert|     Sales|  4100|\n",
      "|         Saif|     Sales|  4100|\n",
      "|      Michael|     Sales|  4600|\n",
      "+-------------+----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df3.sort(df3.salary.desc()).show()\n",
    "df3.sort(df3.salary.desc(),(df3.department)).show()\n",
    "df3.orderBy(df3.salary).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "a67eaf98",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|SPARK_PARTITION_ID()|count|\n",
      "+--------------------+-----+\n",
      "|                   6|    8|\n",
      "|                   7|    2|\n",
      "+--------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import spark_partition_id\n",
    "df3.repartition(8).groupBy(spark_partition_id()).count().show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}