PyPsark Excercise - For Interviews.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4d824f21-c01c-45f2-a865-67023393232a",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+---+-----------+\n|   Name|Age|      Index|\n+-------+---+-----------+\n|  Alice| 30|17179869184|\n|    Bob| 25|42949672960|\n|Charlie| 35|60129542144|\n+-------+---+-----------+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import monotonically_increasing_id\n",
    "\n",
    "# Initialize Spark Session\n",
    "spark = SparkSession.builder.appName(\"IndexColumn\").getOrCreate()\n",
    "\n",
    "# Sample Data\n",
    "data = [(\"Alice\", 30), (\"Bob\", 25), (\"Charlie\", 35)]\n",
    "columns = [\"Name\", \"Age\"]\n",
    "\n",
    "# Create DataFrame\n",
    "df = spark.createDataFrame(data, columns)\n",
    "\n",
    "# Add an Index Column\n",
    "df_with_index = df.withColumn(\"Index\", monotonically_increasing_id())\n",
    "\n",
    "# Show Output\n",
    "df_with_index.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "7b8a0251-31ed-4848-8d83-8e89d1635e43",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+---+-----+\n|   Name|Age|Index|\n+-------+---+-----+\n|  Alice| 30|    1|\n|    Bob| 25|    2|\n|Charlie| 35|    3|\n+-------+---+-----+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.functions import row_number\n",
    "\n",
    "# Define Window Specification\n",
    "windowSpec = Window.orderBy(\"Name\")\n",
    "\n",
    "# Add Index Column\n",
    "df_with_seq_index = df.withColumn(\"Index\", row_number().over(windowSpec))\n",
    "\n",
    "# Show Output\n",
    "df_with_seq_index.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e4d45a6a-3cc6-4514-b51a-518ad7892be6",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n| ID|Category|\n+---+--------+\n|  1|   Apple|\n|  2|  Banana|\n|  3|   Apple|\n|  4|   Other|\n|  5|   Apple|\n|  6|  Banana|\n|  7|   Other|\n|  8|   Other|\n|  9|   Other|\n| 10|   Other|\n+---+--------+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, count, when\n",
    "\n",
    "# Initialize Spark Session\n",
    "spark = SparkSession.builder.appName(\"Top2FrequentValues\").getOrCreate()\n",
    "\n",
    "# Sample Data\n",
    "data = [\n",
    "    (1, \"Apple\"), (2, \"Banana\"), (3, \"Apple\"), (4, \"Orange\"),\n",
    "    (5, \"Apple\"), (6, \"Banana\"), (7, \"Orange\"), (8, \"Grape\"),\n",
    "    (9, \"Grape\"), (10, \"Mango\")\n",
    "]\n",
    "\n",
    "# Create DataFrame\n",
    "df = spark.createDataFrame(data, [\"ID\", \"Category\"])\n",
    "\n",
    "# Count frequency of each category\n",
    "category_counts = df.groupBy(\"Category\").count()\n",
    "\n",
    "# Get the top 2 most frequent categories\n",
    "top_categories = category_counts.orderBy(col(\"count\").desc()).limit(2)\n",
    "\n",
    "# Extract top categories as a list\n",
    "top_categories_list = [row[\"Category\"] for row in top_categories.collect()]\n",
    "\n",
    "# Replace all non-top categories with \"Other\"\n",
    "df_modified = df.withColumn(\n",
    "    \"Category\",\n",
    "    when(col(\"Category\").isin(top_categories_list), col(\"Category\")).otherwise(\"Other\")\n",
    ")\n",
    "\n",
    "# Show the result\n",
    "df_modified.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e880084b-c091-4896-a4b8-7efba4dd434c",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---+-------+\n|name|age|address|\n+----+---+-------+\n|John| 30|     NY|\n|Anna| 40|     LA|\n|Mike| 50|     SF|\n+----+---+-------+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Initialize Spark Session\n",
    "spark = SparkSession.builder.appName(\"RenameColumns\").getOrCreate()\n",
    "\n",
    "# Sample Data\n",
    "data = [(\"John\",30,  \"NY\"), (\"Anna\",40,  \"LA\"), (\"Mike\",50,  \"SF\")]\n",
    "\n",
    "# Create DataFrame with old column names\n",
    "old_columns = [\"col1\", \"col2\", \"col3\"]\n",
    "df = spark.createDataFrame(data, old_columns)\n",
    "\n",
    "# New column names\n",
    "new_columns = [\"name\", \"age\", \"address\"]\n",
    "\n",
    "# Rename columns using old and new column names\n",
    "for old_col, new_col in zip(old_columns, new_columns):\n",
    "    df = df.withColumnRenamed(old_col, new_col)\n",
    "\n",
    "# Show the result\n",
    "df.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f1ab9b1d-1599-484b-9b03-38592e8cc53c",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------------------+----------+-----------+----------+\n| ID|             Values|First_Item|Second_Item|Third_Item|\n+---+-------------------+----------+-----------+----------+\n|  1|   [10, 20, 30, 40]|        10|         20|        30|\n|  2|   [50, 60, 70, 80]|        50|         60|        70|\n|  3|[90, 100, 110, 120]|        90|        100|       110|\n+---+-------------------+----------+-----------+----------+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col\n",
    "\n",
    "# Initialize Spark Session\n",
    "spark = SparkSession.builder.appName(\"ExtractItemsAtPositions\").getOrCreate()\n",
    "\n",
    "# Sample Data: A column containing lists\n",
    "data = [(1, [10, 20, 30, 40]), (2, [50, 60, 70, 80]), (3, [90, 100, 110, 120])]\n",
    "\n",
    "# Create DataFrame\n",
    "df = spark.createDataFrame(data, [\"ID\", \"Values\"])\n",
    "\n",
    "# Extract the first and second items using getItem()\n",
    "df_extracted = df.withColumn(\"First_Item\", col(\"Values\").getItem(0)) \\\n",
    "                 .withColumn(\"Second_Item\", col(\"Values\").getItem(1)).withColumn(\"Third_Item\", col(\"Values\").getItem(2))\n",
    "\n",
    "# Show the result\n",
    "df_extracted.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "17895cc8-70ca-452a-a291-3b240e801722",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+\n| ID| name|\n+---+-----+\n|  1| John|\n|  2|Alice|\n|  3|  Bob|\n|  4|Carol|\n+---+-----+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, upper, substring\n",
    "from pyspark.sql.functions import initcap\n",
    "# Initialize Spark Session\n",
    "spark = SparkSession.builder.appName(\"CapitalizeFirstCharacter\").getOrCreate()\n",
    "\n",
    "# Sample Data\n",
    "data = [(1, \"john\"), (2, \"alice\"), (3, \"bob\"), (4, \"carol\")]\n",
    "\n",
    "# Create DataFrame\n",
    "df = spark.createDataFrame(data, [\"ID\", \"Name\"])\n",
    "\n",
    "# Capitalize the first character of each element in the 'Name' column\n",
    "# df_modified = df.withColumn(\"Name\", \n",
    "#                             upper(substring(col(\"Name\"), 1, 1)) + substring(col(\"Name\"), 2, 100))\n",
    "df_modified = df.withColumn(\"name\", initcap(df[\"name\"]))\n",
    "# Show the result\n",
    "df_modified.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a02afc62-30d3-40e1-a5ad-083da66ac212",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+--------------------+-------------+-------------+-------------+\n| ID|            Sentence|               Words|Word_1_Length|Word_2_Length|Word_3_Length|\n+---+--------------------+--------------------+-------------+-------------+-------------+\n|  1|    PySpark is great|[PySpark, is, great]|            7|            2|            5|\n|  2| Big Data is amazing|[Big, Data, is, a...|            3|            4|            2|\n|  3|Data Engineering ...|[Data, Engineerin...|            4|           11|            5|\n+---+--------------------+--------------------+-------------+-------------+-------------+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, length, split\n",
    "\n",
    "# Initialize Spark Session\n",
    "spark = SparkSession.builder.appName(\"WordLength\").getOrCreate()\n",
    "\n",
    "# Sample Data\n",
    "data = [(1, \"PySpark is great\"), (2, \"Big Data is amazing\"), (3, \"Data Engineering Rocks\")]\n",
    "\n",
    "# Create DataFrame\n",
    "df = spark.createDataFrame(data, [\"ID\", \"Sentence\"])\n",
    "\n",
    "# Split the Sentence into Words\n",
    "df_split = df.withColumn(\"Words\", split(col(\"Sentence\"), \" \"))\n",
    "\n",
    "# Calculate the length of each word in the 'Words' column\n",
    "df_lengths = df_split.withColumn(\"Word_1_Length\", length(col(\"Words\").getItem(0))) \\\n",
    "                     .withColumn(\"Word_2_Length\", length(col(\"Words\").getItem(1))) \\\n",
    "                     .withColumn(\"Word_3_Length\", length(col(\"Words\").getItem(2)))\n",
    "\n",
    "# Show the result\n",
    "df_lengths.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f22c8ed8-bfa3-4322-acbf-6a1630ebe08f",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+----+------------+\n| ID|Value|Diff|Diff_of_Diff|\n+---+-----+----+------------+\n|  1|    5|NULL|        NULL|\n|  2|    8|   3|        NULL|\n|  3|   12|   4|           1|\n|  4|   15|   3|          -1|\n|  5|   20|   5|           2|\n+---+-----+----+------------+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.window import Window\n",
    "\n",
    "# Initialize Spark Session\n",
    "spark = SparkSession.builder.appName(\"DifferenceOfDifferences\").getOrCreate()\n",
    "\n",
    "# Sample Data\n",
    "data = [(1, 5), (2, 8), (3, 12), (4, 15), (5, 20)]\n",
    "\n",
    "# Create DataFrame\n",
    "df = spark.createDataFrame(data, [\"ID\", \"Value\"])\n",
    "\n",
    "# Define a window specification\n",
    "windowSpec = Window.orderBy(\"ID\")\n",
    "\n",
    "# Compute the difference between consecutive values\n",
    "df_diff = df.withColumn(\"Diff\", F.col(\"Value\") - F.lag(\"Value\", 1).over(windowSpec))\n",
    "\n",
    "# Compute the difference of differences\n",
    "df_diff_of_diff = df_diff.withColumn(\"Diff_of_Diff\", F.col(\"Diff\") - F.lag(\"Diff\", 1).over(windowSpec))\n",
    "\n",
    "# Show the result\n",
    "df_diff_of_diff.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5fa53709-b23f-4a51-9fdd-1e054ec0407f",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------+\n| ID|  Word|\n+---+------+\n|  1| Apple|\n|  2|Banana|\n|  4|Grapes|\n+---+------+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import BooleanType\n",
    "\n",
    "# Initialize Spark Session\n",
    "spark = SparkSession.builder.appName(\"FilterWordsWithVowels\").getOrCreate()\n",
    "\n",
    "# Sample Data\n",
    "data = [(1, \"Apple\"), (2, \"Banana\"), (3, \"Sky\"), (4, \"Grapes\"), (5, \"Cat\")]\n",
    "\n",
    "# Create DataFrame\n",
    "df = spark.createDataFrame(data, [\"ID\", \"Word\"])\n",
    "\n",
    "# Define UDF to check if a word contains at least 2 vowels\n",
    "def contains_two_vowels(word):\n",
    "    vowels = \"aeiou\"\n",
    "    word = word.lower()  # Convert to lowercase\n",
    "    vowel_count = sum(1 for char in word if char in vowels)\n",
    "    return vowel_count >= 2\n",
    "\n",
    "# Register the UDF\n",
    "contains_two_vowels_udf = udf(contains_two_vowels, BooleanType())\n",
    "\n",
    "# Filter words that contain at least 2 vowels\n",
    "df_filtered = df.filter(contains_two_vowels_udf(df[\"Word\"]))\n",
    "\n",
    "# Show the result\n",
    "df_filtered.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c826291b-76f2-472c-ab57-90e14a0b0e8c",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+---+\n| ID|  A|  B|\n+---+---+---+\n|  1| 10| 20|\n|  3| 50| 60|\n|  2| 30| 40|\n+---+---+---+\n\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import sum\n",
    "\n",
    "# Initialize Spark session\n",
    "spark = SparkSession.builder.appName(\"PivotDataFrame\").getOrCreate()\n",
    "\n",
    "# Sample Data\n",
    "data = [(1, \"A\", 10), (1, \"B\", 20), (2, \"A\", 30), (2, \"B\", 40), (3, \"A\", 50), (3, \"B\", 60)]\n",
    "\n",
    "# Create DataFrame\n",
    "df = spark.createDataFrame(data, [\"ID\", \"Category\", \"Value\"])\n",
    "\n",
    "# Pivot the DataFrame\n",
    "pivot_df = df.groupBy(\"ID\").pivot(\"Category\").agg(sum(\"Value\"))\n",
    "\n",
    "# Show the pivoted DataFrame\n",
    "pivot_df.show()"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "computePreferences": null,
   "dashboards": [],
   "environmentMetadata": {
    "base_environment": "",
    "client": "1"
   },
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "PyPsark Excercise - For Interviews",
   "widgets": {}
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}