Notebooks/Spark.NET C#/Tweets/01_TweetAnalysis_DataPrep.ipynb

{
  "metadata": {
    "description": "This Spark .NET notebook prepares the tweet analysis tables Mentions and Topics in the TweetAnalysis database.\nIt shows the use of Spark.NET with focusing on:\nHow to use a .NET function (cell 4) in a notebook as a Spark UDF\ncall into SparkSQL\nCreate the Spark tables to be used with other notebooks and even engines",
    "saveOutput": true,
    "language_info": {
      "name": "csharp"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2,
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# TweetAnalysis - Data Prep\n",
        "This Spark .NET notebook prepares the tweet analysis tables Mentions and Topics in the TweetAnalysis database.\n",
        "\n",
        "It shows the use of Spark.NET with focus on:\n",
        "- How to use a .NET function (cell 5) in a notebook as a Spark UDF\n",
        "- How to use some of the Spark functions from .NET (including different ways to reference columns)\n",
        "- call into SparkSQL\n",
        "- Create the Parquet backed Spark tables to be used with other Spark applications and notebooks and even SQL engines\n",
        "\n",
        "**Please replace the `inputfile` path with the location where you placed the Tweet files.** \n",
        "\n",
        "If you would like to use your own tweet data, you can use https://tweetdownload.net and save the result as CSV."
      ],
      "attachments": {}
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## This notebook gives us interactive C#!\n",
        ""
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 16,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/html": "<table><thead><tr><th><i>index</i></th><th>value</th></tr></thead><tbody><tr><td>0</td><td>a</td></tr><tr><td>1</td><td>b</td></tr><tr><td>2</td><td>c</td></tr></tbody></table>"
          },
          "execution_count": 16,
          "metadata": {}
        }
      ],
      "metadata": {},
      "source": [
        "bool display = true; \n",
        "\n",
        "\"a b c\".Split(' ')"
      ],
      "attachments": {}
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Define a C# function `extract_items`\n",
        "\n",
        "`extract_items` returns a list of words from a tweet that were prefixed with the provided `prefix` parameter. \n",
        "\n",
        "This means that you get mentioned tweet handles if you use `prefix:\"@\"` and tweet topics if you use `prefix:\"#\"`.\n",
        "\n",
        "Note that this function can be used now in the notebook."
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "outputs": [],
      "metadata": {},
      "source": [
        "    static IEnumerable<string> extract_items(string tweet, string prefix)\n",
        "    {\n",
        "            return tweet.Split(new char[] { ' ', ',', '.', ':', '!', ';', '\"', '“', ')', '?', '\\'' })\n",
        "                        .Where(x => x.StartsWith(prefix) && x != prefix)\n",
        "                        .Select(x => x.Substring(1));\n",
        "    }"
      ],
      "attachments": {}
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Creating the 'tweets' dataframe\n",
        "\n",
        "This creates the `tweets` dataframe by reading all the tweet files from the specified path (note the file wildcard in the path).\n",
        "\n",
        "Will remove duplicate tweets.\n",
        "\n",
        "**Please replace the `inputfile` path with the location where you placed the Tweet files.** \n",
        "\n",
        "If you would like to use your own tweet data, you can use https://tweetdownload.net and save the result as CSV."
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": "Number of distinct tweets: 17570 - Removed number of duplicates: 150"
          },
          "execution_count": 7,
          "metadata": {}
        }
      ],
      "metadata": {},
      "source": [
        "var inputfile = @\"abfss://<container>@<ADLSGen2Acct>.dfs.core.windows.net/<path>/Tweets/*.csv\";\n",
        "long before_count = 0;\n",
        "\n",
        "var tweets = spark.Read().Schema(\"date STRING, time STRING, author STRING, tweet STRING\").Format(\"csv\").Load(inputfile);\n",
        "if (display) {before_count= tweets.Count();}\n",
        "tweets = tweets.Distinct();\n",
        "if (display) {var after_count = tweets.Count(); Console.WriteLine(\"Number of distinct tweets: \"+after_count.ToString()+\" - Removed number of duplicates: \"+(before_count-after_count).ToString());}"
      ],
      "attachments": {}
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Registering the C# function as Spark UDF\n",
        "The following registers the `extract_items` function as a Spark UDF. \n",
        "\n",
        "Since the `prefix` parameter is a string value that is not taken from a column and registered Spark UDFs need to reference columns, we provide two explicitly named UDFs `extract_mentions` and `extract_topics` that hard-code the `prefix` for the given usage. \n",
        ""
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "outputs": [],
      "metadata": {},
      "source": [
        "Func<Column, Column> extract_mentions = \n",
        "        Udf<string, IEnumerable<string>>((tweet) => extract_items(tweet, \"@\"));\n",
        "        \n",
        "Func<Column, Column> extract_topics = \n",
        "        Udf<string, IEnumerable<string>>((tweet) => extract_items(tweet, \"#\"));"
      ],
      "attachments": {}
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Extracting Mentions and Topics\n",
        "Now we can use the UDFs to extract the mention and topics. Both are represented as string arrays in the form `IEnumerable<string>`.\n",
        "\n",
        "Note that we can refer to the colums either via the generic `Col(colname)` function or by referring to the explicitly named dataframe column accessor `dataframe[colname]`.\n",
        ""
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 9,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": "+----------+-----+---------+--------------------+--------------------------------+\n|      date| time|   author|            mentions|                          topics|\n+----------+-----+---------+--------------------+--------------------------------+\n|18/09/2015|00:01|       iC|         [dahowlett]|                              []|\n|01/09/2015|16:56|       iC|[couchbase, museu...|                              []|\n|16/08/2015|14:12|  etsurow|                  []|[J\u0016qkjcfD�����]|\n|26/04/2015|00:04|       iC|              [lfnw]|                              []|\n|14/02/2015|22:20|       iC|          [Stanford]|                 [ValentinesDay]|\n|11/11/2014|02:48|       iC|[benkepes, jobswo...|                              []|\n|30/07/2014|21:12|       iC|             [wsdot]|                              []|\n|12/07/2014|02:03|       iC|    [xmlgrrl, UMAWG]|              [pbsnewshour, pii]|\n|11/07/2014|04:50|       iC|[Paul_Hofmann, dw...|                              []|\n|16/06/2014|19:04|       iC|                  []|                              []|\n|28/04/2015|10:34|  PulsWeb|       [sebastianbk]|                              []|\n|23/02/2015|09:07|  PulsWeb|       [GUSS_FRANCE]|                     [SQLFamily]|\n|17/10/2014|13:57|  PulsWeb|       [GUSS_FRANCE]|                       [JSS2004]|\n|24/06/2014|15:20|  PulsWeb|          [benjguin]|              [AzureCamp, Azure]|\n|26/06/2013|18:31|  PulsWeb|[GUSS_FRANCE, dba...|                              []|\n|14/03/2013|11:27|  pgeiger|           [PulsWeb]|                              []|\n|03/12/2012|11:08|  PulsWeb|          [dennylee]|            [Hadoop, HDInsigh...|\n|25/06/2012|19:59|  PulsWeb|                  []|               [SQLServer, MSBI]|\n|30/04/2014|20:32|  saveenr|                  []|                     [pshsummit]|\n|16/11/2011|11:29|lllamaboy|[saveenr, Poshoho...|                              []|\n+----------+-----+---------+--------------------+--------------------------------+\nonly showing top 20 rows"
          },
          "execution_count": 9,
          "metadata": {}
        }
      ],
      "metadata": {},
      "source": [
        "var mentionsandtopics = tweets.Select(Col(\"date\"), Col(\"time\"), Col(\"author\")\n",
        "                                    , extract_mentions(tweets[\"tweet\"]).As(\"mentions\")\n",
        "                                    , extract_topics(tweets[\"tweet\"]).As(\"topics\")\n",
        "                );\n",
        "           \n",
        "if (display) {mentionsandtopics.Show();}"
      ],
      "attachments": {}
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Create mentions and topics dataframes\n",
        "\n",
        "We want to pivot the arrays into one row per array item. To avoid the cartesian product between mentions and topics, we create one dataframe each.\n",
        ""
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": "+----------+-----+-------+--------------+\n|      date| time| author|       mention|\n+----------+-----+-------+--------------+\n|18/09/2015|00:01|     iC|     dahowlett|\n|01/09/2015|16:56|     iC|     couchbase|\n|01/09/2015|16:56|     iC|museumofflight|\n|01/09/2015|16:56|     iC|      AWScloud|\n|01/09/2015|16:56|     iC|   googlecloud|\n|26/04/2015|00:04|     iC|          lfnw|\n|14/02/2015|22:20|     iC|      Stanford|\n|11/11/2014|02:48|     iC|      benkepes|\n|11/11/2014|02:48|     iC|     jobsworth|\n|11/11/2014|02:48|     iC|    salesforce|\n|11/11/2014|02:48|     iC|  parkerharris|\n|11/11/2014|02:48|     iC|       Benioff|\n|11/11/2014|02:48|     iC|        fscavo|\n|30/07/2014|21:12|     iC|         wsdot|\n|12/07/2014|02:03|     iC|       xmlgrrl|\n|12/07/2014|02:03|     iC|         UMAWG|\n|11/07/2014|04:50|     iC|  Paul_Hofmann|\n|11/07/2014|04:50|     iC|      dwavesys|\n|28/04/2015|10:34|PulsWeb|   sebastianbk|\n|23/02/2015|09:07|PulsWeb|   GUSS_FRANCE|\n+----------+-----+-------+--------------+\nonly showing top 20 rows"
          },
          "execution_count": 10,
          "metadata": {}
        }
      ],
      "metadata": {},
      "source": [
        "var mentions = mentionsandtopics.Select(Col(\"date\"), Col(\"time\"), Col(\"author\"), Col(\"mentions\").As(\"mention\"))\n",
        "                                .WithColumn(\"mention\", Explode(Col(\"mention\")));\n",
        "if (display) {mentions.Show();}"
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": "+----------+-----+---------------+------------+\n|      date| time|         author|       topic|\n+----------+-----+---------------+------------+\n|09/03/2015|20:38|       SQLCindy|   HDInsight|\n|09/03/2015|20:38|       SQLCindy|       Azure|\n|09/03/2015|20:38|       SQLCindy|        MSBI|\n|09/03/2015|20:38|       SQLCindy|      Hadoop|\n|09/03/2015|20:38|       SQLCindy|    SQLAzure|\n|25/06/2013|02:23|       SQLCindy| BigDataCamp|\n|25/06/2013|02:23|       SQLCindy|   HDInsight|\n|25/06/2013|02:23|       SQLCindy|        MSBI|\n|16/04/2013|14:52|        sqlrnnr|      Hadoop|\n|16/04/2013|14:52|        sqlrnnr|      SSSOLV|\n|18/03/2013|15:53|       SQLCindy|  BigData100|\n|19/02/2013|17:04|       SQLBalls|       MVP13|\n|19/02/2013|17:04|       SQLBalls|         WIT|\n|19/02/2013|17:04|       SQLBalls|     SQLPASS|\n|18/02/2013|20:01|       SQLCindy|       mvp13|\n|18/02/2013|20:01|       SQLCindy|     mvpbuzz|\n|18/02/2013|20:01|       SQLCindy|   sqlserver|\n|11/08/2012|00:12|    sqlagentman|     BigData|\n|11/08/2012|00:12|    sqlagentman|         LOL|\n|27/02/2018|08:37|MikeDoesBigData|SQLKonferenz|\n+----------+-----+---------------+------------+\nonly showing top 20 rows"
          },
          "execution_count": 11,
          "metadata": {}
        }
      ],
      "metadata": {},
      "source": [
        "var topics = mentionsandtopics.Select(Col(\"date\"), Col(\"time\"), Col(\"author\"), Col(\"topics\").As(\"topic\"))\n",
        "                              .WithColumn(\"topic\", Explode(Col(\"topic\")));\n",
        "if (display) {topics.Show();}"
      ],
      "attachments": {}
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Cleaning up the existing Database\n",
        "\n",
        "First we need to drop the tables since Spark requires that a database is empty before we can drop the Database.\n",
        "\n",
        "Then we recreate the database and set the default database context to it.\n",
        ""
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 12,
      "outputs": [],
      "metadata": {},
      "source": [
        "spark.Sql(\"DROP TABLE IF EXISTS BuildTweetAnalysis.Mentions\"); spark.Sql(\"DROP TABLE IF EXISTS BuildTweetAnalysis.Topics\");"
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 13,
      "outputs": [],
      "metadata": {},
      "source": [
        "spark.Sql(\"DROP DATABASE IF EXISTS BuildTweetAnalysis\"); spark.Sql(\"CREATE DATABASE BuildTweetAnalysis\"); \n",
        "spark.Sql(\"USE BuildTweetAnalysis\");"
      ],
      "attachments": {}
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Creating the new tables\n",
        "We create a mentions table from the mentions dataframe and a topics table from the topics dataframe.\n",
        "\n",
        "Since we only have a few hundred KB of data, we want to avoid overpartitioning of the tables into too many files. Thus we repartition the data into a single partition.\n",
        "\n",
        "Note that this simple way of creation will not provide for specifying cluster keys (which you want to specify for large production tables)."
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": "+--------+\n|count(1)|\n+--------+\n|   25246|\n+--------+"
          },
          "execution_count": 14,
          "metadata": {}
        }
      ],
      "metadata": {},
      "source": [
        "mentions.Repartition(1).Write().SaveAsTable(\"Mentions\");\n",
        "if (display) {spark.Sql(\"SELECT COUNT(*) FROM Mentions\").Show();}"
      ],
      "attachments": {}
    },
    {
      "cell_type": "code",
      "execution_count": 15,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": "+--------+\n|count(1)|\n+--------+\n|   14587|\n+--------+"
          },
          "execution_count": 15,
          "metadata": {}
        }
      ],
      "metadata": {},
      "source": [
        "topics.Repartition(1).Write().SaveAsTable(\"BuildTweetAnalysis.Topics\");\n",
        "if (display) {spark.Sql(\"SELECT COUNT(*) FROM BuildTweetAnalysis.Topics\").Show();}"
      ],
      "attachments": {}
    }
  ]
}