pySpark-CheatSheet.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark Basic Architecture and Terminology"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Un applicazione Spark consiste in un **Driver Program** e di un gruppo di **Executors** sul cluster:\n",
    "\n",
    "- Il **driver** è il processo che:\n",
    "\n",
    "    1) esegue il main della Spark Application\n",
    "    \n",
    "    2) crea **SparkContext** che permette di stabilire una comunicazione con il cluster ed i \"resource manager\" al fine di coordinare ed eseguire i **tasks**\n",
    "    \n",
    "    \n",
    "- Gli **executors** sono processi eseguiti sui nodi **worker** del cluster che sono responsabili di eseguire i **task** che il driver ha assegnato a loro.\n",
    "\n",
    "\n",
    "- Il **cluster manager** (*Mesos* o *YARN*) è responsabile dell'allocazione delle risorse fisiche per la spark application\n",
    "\n",
    "<img src=\"https://spark.apache.org/docs/latest/img/cluster-overview.png\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ogni applicazione Spark necessita di un **entry point** che le consenta di comunicare con basi di dati ed eseguire determinate operazioni come la lettura e la scrittura di dati. Da Spark 2.x questo è entry point è la **SparkSession** che sostituisce i precedenti *SparkContext, SQLContext* e HiveContext di Spark 1.x."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark SQL basic example\") \\\n",
    "    .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Lettura dei dati"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# method 1 for reading a CSV file\n",
    "df = spark.read.csv(file_path, header=True)\n",
    "\n",
    "# method 2 for reading a CSV file\n",
    "df = spark.\\ read \\\n",
    "            .format(csv_plugin) \\\n",
    "            .options(header='true', inferSchema='true') \\\n",
    "            .load(file_path)\n",
    "\n",
    "# Reading a json file\n",
    "df = spark.read.json(json_file_path)\n",
    "\n",
    "# Reading a text file\n",
    "df = spark.read.text(text_file_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creazione dei DataFrame"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 1 metodo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Row\n",
    "\n",
    "# populate two rows with random values\n",
    "f1 = Row(original_title='Eroica', budget='13393950', year=1992)\n",
    "f2 = Row(original_title='Night World', budget='1255930', year=1998)\n",
    "\n",
    "# store the two rows in an array and pass it to Spark\n",
    "films = [f1, f2]\n",
    "df = spark.createDataFrame(films)\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 2 metodo: creazione a partire da RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StringType, StructField, StructType, IntegerType\n",
    "\n",
    "rdd = spark.textFile(csv_file_path)\n",
    "schema = StructType([\n",
    "        StructField(\"first_name\", StringType(), True),\n",
    "        StructField(\"last_name\", StringType(), True),\n",
    "        StructField(\"age\", IntegerType(), True)\n",
    "    ])\n",
    "   \n",
    "df = spark.createDataFrame(rdd, schema)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 3 metodo: creazione a partire da query SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#eseguire una query sql direttamente su un file parquet\n",
    "df = spark.sql(\"SELECT * FROM parquet.`examples/src/main/resources/users.parquet`\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lettura DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head(5)\n",
    "data.select(\"text\",\"date\").show()\n",
    "data.printSchema()\n",
    "data.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Modifiche DataFrame"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 1) Creare una colonna"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.types import StringType\n",
    "\n",
    "# Create a column with the default value = 'xyz'\n",
    "df = df.withColumn('new_column', F.lit('xyz'))\n",
    "\n",
    "# Create a column with default value as null\n",
    "df = df.withColumn('new_column', F.lit(None).cast(StringType()))\n",
    "\n",
    "# Create a column using an existing column\n",
    "df = df.withColumn('new_column', 1.4 * F.col('existing_column'))\n",
    "\n",
    "# Another example using the MovieLens database\n",
    "df = df.withColumn('test_col3', F.when(F.col('avg_ratings') < 7, 'OK')\\\n",
    "                                 .when(F.col('avg_ratings') < 8, 'Good')\\\n",
    "                                 .otherwise('Great')).show()\n",
    "\n",
    "# Create a column using a UDF\n",
    "def categorize(val):\n",
    "    if val < 150: \n",
    "        return 'bucket_1'\n",
    "    else:\n",
    "        return 'bucket_2'\n",
    "    \n",
    "my_udf = F.udf(categorize, StringType())\n",
    "\n",
    "df = df.withColumn('new_column', categorize('existing_column'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 2) Eliminare una colonna"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Remove single column\n",
    "df.drop('this_column')\n",
    "\n",
    "# Remove multiple columns in a go\n",
    "drop_columns = ['this_column', 'that_column']\n",
    "df.select([col for col in df.columns if column not in drop_columns])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 3) Rinominare una colonna"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Changing column name with withColumnRenamed feature\n",
    "df = df.withColumnRenamed('existing_column_name', 'new_column_name')\n",
    "\n",
    "# Changing column with selectExpr (you'll have to select all the columns here)\n",
    "df = df.selectExpr(\"existing_column_name AS existing_1\", \"new_column_name AS new_1\")\n",
    "\n",
    "df = df.select(col(\"existing_column_name\").alias(\"existing_1\"), col(\"new_column_name\").alias(\"new_1\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 4) Sintassi SQL in Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'spark' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-1-f18b4310a071>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0mdf\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mspark\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msql\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"SELECT * FROM parquet.`examples/src/main/resources/users.parquet`\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;31mNameError\u001b[0m: name 'spark' is not defined"
     ]
    }
   ],
   "source": [
    "# Creates a temporary view using the DataFrame\n",
    "df.createOrReplaceTempView(\"people\")\n",
    "\n",
    "# SQL statements can be run by using the sql methods provided by spark\n",
    "query = \"SELECT name FROM people WHERE age BETWEEN 13 AND 19\"\n",
    "teenagerNamesDF = spark.sql(query)\n",
    "teenagerNamesDF.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.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}