connector/examples/AccumuloSparkConnector.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Example Microsoft MASC Demo Notebook\n",
    "\n",
    "## Tweet Based Sentiment Analysis\n",
    "This example uses the Sentiment140 dataset to demonstrate simple tweet-based sentiment analysis\n",
    "\n",
    "### Requirements\n",
    "- Python version should be >= 3\n",
    "- Jupyter Notebook should run using Java SDK version == 1.8\n",
    "- Accumulo version should be >= 2.0.0\n",
    "\n",
    "### Prerequisites\n",
    "- Download and extract the dataset (available here: http://help.sentiment140.com/for-students)\n",
    "  - In this example we have already extracted and pushed the training data file to HDFS\n",
    "- Build datasource and iterator JARs or pull from Maven Central Repository\n",
    "  - Datasource package on Maven Central Repository is specified below\n",
    "  - Iterator JAR is deployed to all Accumulo nodes then Accumulo cluster is restarted\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import base64\n",
    "from configparser import ConfigParser\n",
    "import os\n",
    "\n",
    "from mleap import pyspark\n",
    "from mleap.pyspark.spark_support import SimpleSparkSerializer\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.feature import CountVectorizer, RegexTokenizer\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.sql import functions as F"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define dependencies and configuration variables\n",
    "DATASOURCE_PKG = 'com.microsoft.masc:microsoft-accumulo-spark-datasource:1.0.3'\n",
    "SPARK_AVRO_PKG = 'org.apache.spark:spark-avro_2.11:2.4.3'\n",
    "SPARK_MLEAP_PKG = 'ml.combust.mleap:mleap-spark_2.11:0.13.0'\n",
    "\n",
    "ACCUMULO_PROPS = '/opt/muchos/install/accumulo-2.0.0/conf/accumulo-client.properties'\n",
    "DATA_FILE = 'hdfs:///data/training.1600000.processed.noemoticon.csv'\n",
    "MLEAP_BUNDLE = 'hdfs:///tmp/sentiment140.lr.zip'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define helper functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def convert_label(df):\n",
    "    \"\"\"Convert label to binary integer\"\"\"\n",
    "    return (df\n",
    "            .filter(df['label'] != '2')\n",
    "            .withColumn('label', (df['label'].cast(IntegerType()) / F.lit(4)).cast(IntegerType())))\n",
    "\n",
    "def get_data(spark, data_file):\n",
    "    \"\"\"Read Sentiment140 data from file\"\"\"\n",
    "    df = (spark.read\n",
    "          .csv(data_file)\n",
    "          .withColumn('label', F.col('_c0').cast(IntegerType()))\n",
    "          .withColumnRenamed('_c1', 'id')\n",
    "          .withColumnRenamed('_c2', 'timestamp')\n",
    "          .withColumnRenamed('_c3', 'query')\n",
    "          .withColumnRenamed('_c4', 'user')\n",
    "          .withColumnRenamed('_c5', 'text')\n",
    "          .drop('_c0'))\n",
    "\n",
    "    return df\n",
    "\n",
    "def get_properties(properties_file):\n",
    "    \"\"\"Read Accumulo client properties file\"\"\"\n",
    "    config = ConfigParser()\n",
    "    with open(properties_file) as stream:\n",
    "        config.read_string(\"[top]\\n\" + stream.read())\n",
    "    return dict(config['top'])\n",
    "\n",
    "def get_model_string(bundle, model, df):\n",
    "    \"\"\"Get base64 encoded string representation of MLeap PipelineModel\"\"\"\n",
    "    if os.path.exists(bundle):\n",
    "        os.remove(bundle)\n",
    "    model.serializeToBundle('jar:file:{}'.format(bundle), model.transform(df))\n",
    "\n",
    "    with open(bundle, mode='rb') as file:\n",
    "        model_string = base64.b64encode(file.read()).decode('utf-8')\n",
    "\n",
    "    return model_string"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read Data\n",
    "Reading subset of twitter data used in sentiment analysis benchmark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start Spark session\n",
    "spark = (SparkSession\n",
    "         .builder\n",
    "         .appName('DataSourceTest')\n",
    "         .config('spark.jars.packages', ','.join([DATASOURCE_PKG, SPARK_AVRO_PKG, SPARK_MLEAP_PKG]))\n",
    "         .getOrCreate())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>id</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>query</th>\n",
       "      <th>user</th>\n",
       "      <th>text</th>\n",
       "      <th>label</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <td>0</td>\n",
       "      <td>1467810369</td>\n",
       "      <td>Mon Apr 06 22:19:45 PDT 2009</td>\n",
       "      <td>NO_QUERY</td>\n",
       "      <td>_TheSpecialOne_</td>\n",
       "      <td>@switchfoot http://twitpic.com/2y1zl - Awww, t...</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>1</td>\n",
       "      <td>1467810672</td>\n",
       "      <td>Mon Apr 06 22:19:49 PDT 2009</td>\n",
       "      <td>NO_QUERY</td>\n",
       "      <td>scotthamilton</td>\n",
       "      <td>is upset that he can't update his Facebook by ...</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>2</td>\n",
       "      <td>1467810917</td>\n",
       "      <td>Mon Apr 06 22:19:53 PDT 2009</td>\n",
       "      <td>NO_QUERY</td>\n",
       "      <td>mattycus</td>\n",
       "      <td>@Kenichan I dived many times for the ball. Man...</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>3</td>\n",
       "      <td>1467811184</td>\n",
       "      <td>Mon Apr 06 22:19:57 PDT 2009</td>\n",
       "      <td>NO_QUERY</td>\n",
       "      <td>ElleCTF</td>\n",
       "      <td>my whole body feels itchy and like its on fire</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>4</td>\n",
       "      <td>1467811193</td>\n",
       "      <td>Mon Apr 06 22:19:57 PDT 2009</td>\n",
       "      <td>NO_QUERY</td>\n",
       "      <td>Karoli</td>\n",
       "      <td>@nationwideclass no, it's not behaving at all....</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           id                     timestamp     query             user  \\\n",
       "0  1467810369  Mon Apr 06 22:19:45 PDT 2009  NO_QUERY  _TheSpecialOne_   \n",
       "1  1467810672  Mon Apr 06 22:19:49 PDT 2009  NO_QUERY    scotthamilton   \n",
       "2  1467810917  Mon Apr 06 22:19:53 PDT 2009  NO_QUERY         mattycus   \n",
       "3  1467811184  Mon Apr 06 22:19:57 PDT 2009  NO_QUERY          ElleCTF   \n",
       "4  1467811193  Mon Apr 06 22:19:57 PDT 2009  NO_QUERY           Karoli   \n",
       "\n",
       "                                                text  label  \n",
       "0  @switchfoot http://twitpic.com/2y1zl - Awww, t...      0  \n",
       "1  is upset that he can't update his Facebook by ...      1  \n",
       "2  @Kenichan I dived many times for the ball. Man...      0  \n",
       "3    my whole body feels itchy and like its on fire       1  \n",
       "4  @nationwideclass no, it's not behaving at all....      0  "
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = get_data(spark=spark, data_file=DATA_FILE)\n",
    "df.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Train Logistic Regression Spark ML Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define pipeline of feature engineering and model training\n",
    "tokenizer = RegexTokenizer(gaps=False, pattern='\\\\p{L}+', inputCol='text', outputCol='words')\n",
    "vectorizer = CountVectorizer(inputCol='words', outputCol='features')\n",
    "lr = LogisticRegression(maxIter=1, regParam=0.2, elasticNetParam=0)\n",
    "\n",
    "pipeline = Pipeline(stages=[tokenizer, vectorizer, lr])\n",
    "\n",
    "# Fit Model\n",
    "model = pipeline.fit(df)\n",
    "\n",
    "# Export Model\n",
    "model_string = get_model_string(bundle=MLEAP_BUNDLE, model=model, df=df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Save Data to Accumulo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "properties = get_properties(ACCUMULO_PROPS)\n",
    "# Define Accumulo table where data will be written\n",
    "properties['table'] = 'demo_table'\n",
    "# Identify column to use as the key for Accumulo rows\n",
    "properties['rowkey'] = 'id'\n",
    "\n",
    "(df.write\n",
    "   .format(\"com.microsoft.accumulo\")\n",
    "   .options(**properties)\n",
    "   .save())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Excecute server-side inference and filtering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>user</th>\n",
       "      <th>text</th>\n",
       "      <th>prediction</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <td>0</td>\n",
       "      <td>scotthamilton</td>\n",
       "      <td>is upset that he can't update his Facebook by ...</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>1</td>\n",
       "      <td>ElleCTF</td>\n",
       "      <td>my whole body feels itchy and like its on fire</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>2</td>\n",
       "      <td>joy_wolf</td>\n",
       "      <td>@Kwesidei not the whole crew</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>3</td>\n",
       "      <td>coZZ</td>\n",
       "      <td>@LOLTrish hey  long time no see! Yes.. Rains a...</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>4</td>\n",
       "      <td>mimismo</td>\n",
       "      <td>@twittera que me muera ?</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>5</td>\n",
       "      <td>pardonlauren</td>\n",
       "      <td>I just re-pierced my ears</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>6</td>\n",
       "      <td>robrobbierobert</td>\n",
       "      <td>@octolinz16 It it counts, idk why I did either...</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>7</td>\n",
       "      <td>HairByJess</td>\n",
       "      <td>@iamjazzyfizzle I wish I got to watch it with ...</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>8</td>\n",
       "      <td>armotley</td>\n",
       "      <td>about to file taxes</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>9</td>\n",
       "      <td>gi_gi_bee</td>\n",
       "      <td>@FakerPattyPattz Oh dear. Were you drinking ou...</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>10</td>\n",
       "      <td>swinspeedx</td>\n",
       "      <td>one of my friend called me, and asked to meet ...</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>11</td>\n",
       "      <td>viJILLante</td>\n",
       "      <td>this week is not going as i had hoped</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>12</td>\n",
       "      <td>ChicagoCubbie</td>\n",
       "      <td>I hate when I have to call and wake people up</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>13</td>\n",
       "      <td>gagoo</td>\n",
       "      <td>im sad now  Miss.Lilly</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>14</td>\n",
       "      <td>BaptisteTheFool</td>\n",
       "      <td>Meh... Almost Lover is the exception... this t...</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "               user                                               text  \\\n",
       "0     scotthamilton  is upset that he can't update his Facebook by ...   \n",
       "1           ElleCTF    my whole body feels itchy and like its on fire    \n",
       "2          joy_wolf                      @Kwesidei not the whole crew    \n",
       "3              coZZ  @LOLTrish hey  long time no see! Yes.. Rains a...   \n",
       "4           mimismo                          @twittera que me muera ?    \n",
       "5      pardonlauren                         I just re-pierced my ears    \n",
       "6   robrobbierobert  @octolinz16 It it counts, idk why I did either...   \n",
       "7        HairByJess  @iamjazzyfizzle I wish I got to watch it with ...   \n",
       "8          armotley                               about to file taxes    \n",
       "9         gi_gi_bee  @FakerPattyPattz Oh dear. Were you drinking ou...   \n",
       "10       swinspeedx  one of my friend called me, and asked to meet ...   \n",
       "11       viJILLante             this week is not going as i had hoped    \n",
       "12    ChicagoCubbie     I hate when I have to call and wake people up    \n",
       "13            gagoo                             im sad now  Miss.Lilly   \n",
       "14  BaptisteTheFool  Meh... Almost Lover is the exception... this t...   \n",
       "\n",
       "    prediction  \n",
       "0          1.0  \n",
       "1          1.0  \n",
       "2          1.0  \n",
       "3          1.0  \n",
       "4          1.0  \n",
       "5          1.0  \n",
       "6          1.0  \n",
       "7          1.0  \n",
       "8          1.0  \n",
       "9          1.0  \n",
       "10         1.0  \n",
       "11         1.0  \n",
       "12         1.0  \n",
       "13         1.0  \n",
       "14         1.0  "
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Define model to use\n",
    "properties['mleap'] = model_string\n",
    "# Define filtering based on model output\n",
    "properties['mleapfilter'] = '${prediction > .9}'\n",
    "# Remove id column when reading as this is populated from the Accumulo key\n",
    "schema = df.drop('id').schema\n",
    "\n",
    "pred = (spark\n",
    "        .read\n",
    "        .format(\"com.microsoft.accumulo\")\n",
    "        .options(**properties)\n",
    "        .schema(schema)\n",
    "        .load())\n",
    "\n",
    "# Define column level filtering\n",
    "pred.select(['user', 'text', 'prediction']).limit(15).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  }
 ],
 "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}