14 - Introduction to Spark SQL in Python.ipynb

{
 "cells": [
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "1 - Pyspark SQL\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Create a SQL table from a dataframe\n",
    "\n",
    "# Load trainsched.txt\n",
    "df = spark.read.csv(\"trainsched.txt\", header=True)\n",
    "\n",
    "# Create temporary table called table1\n",
    "df.createOrReplaceTempView('table1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Determine the column names of a table\n",
    "\n",
    "# Inspect the columns in the table df\n",
    "spark.sql(\"DESCRIBE schedule\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Running sums using window function SQL\n",
    "\n",
    "# Add col running_total that sums diff_min col in each group\n",
    "query = \"\"\"\n",
    "SELECT train_id, station, time, diff_min,\n",
    "SUM(diff_min) OVER (PARTITION BY train_id ORDER BY time) AS running_total\n",
    "FROM schedule\n",
    "\"\"\"\n",
    "\n",
    "# Run the query and display the result\n",
    "spark.sql(query).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Fix the broken query\n",
    "\n",
    "query = \"\"\"\n",
    "SELECT \n",
    "ROW_NUMBER() OVER (ORDER BY time) AS row,\n",
    "train_id, \n",
    "station, \n",
    "time, \n",
    "LEAD(time,1) OVER (ORDER BY time) AS time_next \n",
    "FROM schedule\n",
    "\"\"\"\n",
    "spark.sql(query).show()\n",
    "\n",
    "# Give the number of the bad row\n",
    "bad_row = 7\n",
    "\n",
    "# Provide the missing clause\n",
    "clause = \"PARTITION BY train_id\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Aggregation, step by step\n",
    "\n",
    "# Give the identical result in each command\n",
    "spark.sql('SELECT train_id, MIN(time) AS start FROM schedule GROUP BY train_id').show()\n",
    "df.groupBy('train_id').agg({'time':'min'}).withColumnRenamed('min(time)', 'start').show()\n",
    "\n",
    "# Print the second column of the result\n",
    "spark.sql('SELECT train_id, MIN(time), MAX(time) FROM schedule GROUP BY train_id').show()\n",
    "result = df.groupBy('train_id').agg({'time':'min', 'time':'max'})\n",
    "result.show()\n",
    "print(result.columns[1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Aggregating the same column twice\n",
    "\n",
    "## from pyspark.sql.functions import min, max, col\n",
    "## expr = [min(col(\"time\")).alias('start'), max(col(\"time\")).alias('end')]\n",
    "## dot_df = df.groupBy(\"train_id\").agg(*expr)\n",
    "## dot_df.show()\n",
    "\n",
    "\n",
    "# Write a SQL query giving a result identical to dot_df\n",
    "query = \"SELECT train_id, MIN(time) AS start, MAX(time) AS end FROM schedule GROUP BY train_id\"\n",
    "sql_df = spark.sql(query)\n",
    "sql_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Aggregate dot SQL\n",
    "\n",
    "# Obtain the identical result using dot notation \n",
    "dot_df = df.withColumn('time_next', lead('time', 1)\n",
    "        .over(Window.partitionBy('train_id')\n",
    "        .orderBy('time')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Convert window function from dot notation to SQL\n",
    "\n",
    "# Create a SQL query to obtain an identical result to dot_df\n",
    "query = \"\"\"\n",
    "SELECT *, \n",
    "(UNIX_TIMESTAMP(LEAD(time, 1) OVER (PARTITION BY train_id ORDER BY time),'H:m') \n",
    " - UNIX_TIMESTAMP(time, 'H:m'))/60 AS diff_min \n",
    "FROM schedule \n",
    "\"\"\"\n",
    "\n",
    "sql_df = spark.sql(query)\n",
    "sql_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "2 - Using window function sql for natural language processing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Loading a dataframe from a parquet file\n",
    "\n",
    "# Load the dataframe\n",
    "df = spark.read.load('sherlock_sentences.parquet')\n",
    "\n",
    "# Filter and show the first 5 rows\n",
    "df.where('id > 70').show(5, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Split and explode a text column\n",
    "\n",
    "# Split the clause column into a column called words \n",
    "split_df = clauses_df.select(split('clause', ' ').alias('words'))\n",
    "split_df.show(5, truncate=False)\n",
    "\n",
    "# Explode the words column into a column called word \n",
    "exploded_df = split_df.select(explode('words').alias('word'))\n",
    "exploded_df.show(10)\n",
    "\n",
    "# Count the resulting number of rows in exploded_df\n",
    "print(\"\\nNumber of rows: \", exploded_df.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Creating context window feature data\n",
    "\n",
    "# Word for each row, previous two and subsequent two words\n",
    "query = \"\"\"\n",
    "SELECT\n",
    "part,\n",
    "LAG(word, 2) OVER(PARTITION BY part ORDER BY id) AS w1,\n",
    "LAG(word, 1) OVER(PARTITION BY part ORDER BY id) AS w2,\n",
    "word AS w3,\n",
    "LEAD(word, 1) OVER(PARTITION BY part ORDER BY id) AS w4,\n",
    "LEAD(word, 2) OVER(PARTITION BY part ORDER BY id) AS w5\n",
    "FROM text\n",
    "\"\"\"\n",
    "\n",
    "spark.sql(query).where(\"part = 12\").show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Repartitioning the data\n",
    "\n",
    "# Repartition text_df into 12 partitions on 'chapter' column\n",
    "repart_df = text_df.repartition(12, 'chapter')\n",
    "\n",
    "# Prove that repart_df has 12 partitions\n",
    "repart_df.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Finding common word sequences\n",
    "\n",
    "# Find the top 10 sequences of five words\n",
    "query = \"\"\"\n",
    "SELECT w1, w2, w3, w4, w5, COUNT(*) AS count FROM (\n",
    "   SELECT word AS w1,\n",
    "   LEAD(word, 1) OVER(PARTITION BY part ORDER BY id) AS w2,\n",
    "   LEAD(word, 2) OVER(PARTITION BY part ORDER BY id) AS w3,\n",
    "   LEAD(word, 3) OVER(PARTITION BY part ORDER BY id) AS w4,\n",
    "   LEAD(word, 4) OVER(PARTITION BY part ORDER BY id) AS W5\n",
    "   FROM text\n",
    ")\n",
    "GROUP BY w1, w2, w3, w4, w5\n",
    "ORDER BY count DESC\n",
    "LIMIT 10 \"\"\"\n",
    "df = spark.sql(query)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Unique 5-tuples in sorted order\n",
    "\n",
    "# Unique 5-tuples sorted in descending order\n",
    "spark.sql(\"\"\"\n",
    "SELECT DISTINCT w1, w2, w3, w4, w5 FROM (\n",
    "   SELECT word AS w1,\n",
    "   LEAD(word,1) OVER(PARTITION BY part ORDER BY id ) AS w2,\n",
    "   LEAD(word,2) OVER(PARTITION BY part ORDER BY id ) AS w3,\n",
    "   LEAD(word,3) OVER(PARTITION BY part ORDER BY id ) AS w4,\n",
    "   LEAD(word,4) OVER(PARTITION BY part ORDER BY id ) AS w5\n",
    "   FROM text\n",
    ")\n",
    "ORDER BY w1 DESC, w2 DESC, w3 DESC, w4 DESC, w5 DESC \n",
    "LIMIT 10\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Most frequent 3-tuples per chapter\n",
    "\n",
    "query = \"\"\"\n",
    "SELECT chapter, w1, w2, w3, count FROM\n",
    "(\n",
    "  SELECT\n",
    "  chapter,\n",
    "  ROW_NUMBER() OVER (PARTITION BY chapter ORDER BY count DESC) AS row,\n",
    "  w1, w2, w3, count\n",
    "  FROM ( %s )\n",
    ")\n",
    "WHERE row = 1\n",
    "ORDER BY chapter ASC\n",
    "\"\"\" % subquery\n",
    "\n",
    "spark.sql(query).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "3 - Caching, Logging, and the Spark UI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Practicing caching: part 1\n",
    "# Unpersists df1 and df2 and initializes a timer\n",
    "prep(df1, df2) \n",
    "\n",
    "# Cache df1\n",
    "df1.cache()\n",
    "\n",
    "# Run actions on both dataframes\n",
    "run(df1, \"df1_1st\") \n",
    "run(df1, \"df1_2nd\")\n",
    "run(df2, \"df2_1st\")\n",
    "run(df2, \"df2_2nd\", elapsed=True)\n",
    "\n",
    "# Prove df1 is cached\n",
    "print(df1.is_cached)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Practicing caching: the SQL\n",
    "\n",
    "# Unpersist df1 and df2 and initializes a timer\n",
    "prep(df1, df2) \n",
    "\n",
    "# Persist df2 using memory and disk storage level \n",
    "df2.persist(storageLevel=pyspark.StorageLevel.MEMORY_AND_DISK)\n",
    "\n",
    "# Run actions both dataframes\n",
    "run(df1, \"df1_1st\") \n",
    "run(df1, \"df1_2nd\") \n",
    "run(df2, \"df2_1st\") \n",
    "run(df2, \"df2_2nd\", elapsed=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Caching and uncaching tables\n",
    "\n",
    "# List the tables\n",
    "print(\"Tables:\\n\", spark.catalog.listTables())\n",
    "\n",
    "# Cache table1 and Confirm that it is cached\n",
    "spark.catalog.isCached('table1')\n",
    "print(\"table1 is cached: \", spark.catalog.cacheTable('table1'))\n",
    "\n",
    "# Uncache table1 and confirm that it is uncached\n",
    "spark.catalog.uncacheTable('table1')\n",
    "print(\"table1 is cached: \", spark.catalog.isCached('table1'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Practice logging\n",
    "\n",
    "# Log columns of text_df as debug message\n",
    "logging.debug(\"text_df columns: %s\", text_df.columns)\n",
    "\n",
    "# Log whether table1 is cached as info message\n",
    "logging.info(\"table1 is cached: %s\", spark.catalog.isCached(tableName=\"table1\"))\n",
    "\n",
    "# Log first row of text_df as warning message\n",
    "logging.warning(\"The first row of text_df:\\n %s\", text_df.first())\n",
    "\n",
    "# Log selected columns of text_df as error message\n",
    "logging.error(\"Selected columns: %s\", text_df.select(\"id\", \"word\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Practice logging 2\n",
    "\n",
    "# Uncomment the statements that do NOT trigger text_df\n",
    "logging.debug(\"text_df columns: %s\", text_df.columns)\n",
    "logging.info(\"table1 is cached: %s\", spark.catalog.isCached(tableName=\"table1\"))\n",
    "# logging.warning(\"The first row of text_df: %s\", text_df.first())\n",
    "logging.error(\"Selected columns: %s\", text_df.select(\"id\", \"word\"))\n",
    "logging.info(\"Tables: %s\", spark.sql(\"show tables\").collect())\n",
    "logging.debug(\"First row: %s\", spark.sql(\"SELECT * FROM table1 limit 1\"))\n",
    "# logging.debug(\"Count: %s\", spark.sql(\"SELECT COUNT(*) AS count FROM table1\").collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Practice query plans\n",
    "\n",
    "# Run explain on text_df\n",
    "text_df.explain()\n",
    "\n",
    "# Run explain on \"SELECT COUNT(*) AS count FROM table1\" \n",
    "spark.sql(\"SELECT COUNT(*) AS count FROM table1\").explain()\n",
    "\n",
    "# Run explain on \"SELECT COUNT(DISTINCT word) AS words FROM table1\"\n",
    "spark.sql(\"SELECT COUNT(DISTINCT word) as words FROM table1\").explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Practice reading query plans 2\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "4 - Text classification"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Practicing creating a UDF\n",
    "\n",
    "# Returns true if the value is a nonempty vector\n",
    "nonempty_udf = udf(lambda x:  \n",
    "    True if (x and hasattr(x, \"toArray\") and x.numNonzeros())\n",
    "    else False, BooleanType())\n",
    "\n",
    "# Returns first element of the array as string\n",
    "s_udf = udf(lambda x: str(x[0]) if (x and type(x) is list and len(x) > 0)\n",
    "    else '', StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Practicing array column\n",
    "\n",
    "# Show the rows where doc contains the item '5'\n",
    "df_before.where(array_contains('doc', '5')).show()\n",
    "\n",
    "# UDF removes items in TRIVIAL_TOKENS from array\n",
    "rm_trivial_udf = udf(lambda x:\n",
    "                     list(set(x) - TRIVIAL_TOKENS) if x\n",
    "                     else x,\n",
    "                     ArrayType(StringType()))\n",
    "\n",
    "# Remove trivial tokens from 'in' and 'out' columns of df2\n",
    "df_after = df_before.withColumn('in', rm_trivial_udf('in'))\\\n",
    "                    .withColumn('out', rm_trivial_udf('out'))\n",
    "\n",
    "# Show the rows of df_after where doc contains the item '5'\n",
    "df_after.where(array_contains('doc','5')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Creating a UDF for vector data\n",
    "\n",
    "# Selects the first element of a vector column\n",
    "first_udf = udf(lambda x:\n",
    "            float(x.indices[0]) \n",
    "            if (x and hasattr(x, \"toArray\") and x.numNonzeros())\n",
    "            else 0.0,\n",
    "            FloatType())\n",
    "\n",
    "# Apply first_udf to the output column\n",
    "df.select(first_udf(\"output\").alias(\"result\")).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Applying a UDF to vector data\n",
    "\n",
    "# Add label by applying the get_first_udf to output column\n",
    "df_new = df.withColumn('label', get_first_udf('output'))\n",
    "\n",
    "# Show the first five rows \n",
    "df_new.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Transforming text to vector format\n",
    "\n",
    "# Transform df using model\n",
    "result = model.transform(df.withColumnRenamed('in', 'words'))\\\n",
    "        .withColumnRenamed('words', 'in')\\\n",
    "        .withColumnRenamed('vec', 'invec')\n",
    "result.drop('sentence').show(3, False)\n",
    "\n",
    "# Add a column based on the out column called outvec\n",
    "result = model.transform(result.withColumnRenamed('out', 'words'))\\\n",
    "        .withColumnRenamed('words', 'out')\\\n",
    "        .withColumnRenamed('vec', 'outvec')\n",
    "result.select('invec', 'outvec').show(3, False)\t"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## label data\n",
    "\n",
    "# Import the lit function\n",
    "from pyspark.sql.functions import lit\n",
    "\n",
    "# Select the rows where endword is 'him' and label 1\n",
    "df_pos = df.where(\"endword = 'him'\")\\\n",
    "           .withColumn('label', lit(1))\n",
    "\n",
    "# Select the rows where endword is not 'him' and label 0\n",
    "df_neg = df.where(\"endword <> 'him'\")\\\n",
    "           .withColumn('label', lit(0))\n",
    "\n",
    "# Union pos and neg in equal number\n",
    "df_examples = df_pos.union(df_neg.limit(df_pos.count()))\n",
    "print(\"Number of examples: \", df_examples.count())\n",
    "df_examples.where(\"endword <> 'him'\").sample(False, .1, 42).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Split the data\n",
    "\n",
    "# Split the examples into train and test, use 80/20 split\n",
    "df_trainset, df_testset = df_examples.randomSplit([.8, .2], 42)\n",
    "\n",
    "# Print the number of training examples\n",
    "print(\"Number training: \", df_trainset.count())\n",
    "\n",
    "# Print the number of test examples\n",
    "print(\"Number test: \", df_testset.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Train the classifier\n",
    "\n",
    "# Import the logistic regression classifier\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "# Instantiate logistic setting elasticnet to 0.0\n",
    "logistic = LogisticRegression(maxIter=100, regParam=0.4, elasticNetParam=0.0)\n",
    "\n",
    "# Train the logistic classifer on the trainset\n",
    "df_fitted = logistic.fit(df_trainset)\n",
    "\n",
    "# Print the number of training iterations\n",
    "print(\"Training iterations: \", df_fitted.summary.totalIterations)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Evaluate the classifier\n",
    "\n",
    "# Score the model on test data\n",
    "testSummary = df_fitted.evaluate(df_testset)\n",
    "\n",
    "# Print the AUC metric\n",
    "print(\"\\ntest AUC: %.3f\" % testSummary.areaUnderROC)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Predict test data\n",
    "\n",
    "# Apply the model to the test data\n",
    "predictions = df_fitted.transform(df_testset).select(fields)\n",
    "\n",
    "# Print incorrect if prediction does not match label\n",
    "for x in predictions.take(8):\n",
    "    print()\n",
    "    if x.label != int(x.prediction):\n",
    "        print(\"INCORRECT ==> \")\n",
    "    for y in fields:\n",
    "        print(y,\":\", x[y])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "tf-gpu-2.0",
   "language": "python",
   "name": "tf_gpu2"
  },
  "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}