predict.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "dd5bd742-a154-4a55-a46f-a7ea3cafa129",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.ml.pipeline import PipelineModel\n",
    "from pyspark.sql.dataframe import DataFrame\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.sql.functions import lower, when, col, udf, split, lit, format_string\n",
    "from dotenv import load_dotenv\n",
    "import os\n",
    "load_dotenv()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b891117c-0759-4a5d-94fc-189fe5bba57a",
   "metadata": {},
   "source": [
    "## Constants"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "b06223cf-9b18-4747-a4cb-0bac6216f341",
   "metadata": {},
   "outputs": [],
   "source": [
    "TRAINING_FILE = os.getenv(\"TRAINING_FILE\",\"dataset/dataset.csv\")\n",
    "SPARK_MASTER = os.getenv(\"SPARK_MASTER\", \"spark://gpu3.esw:7077\")\n",
    "KAFKA_SERVER = os.getenv(\"KAFKA_SERVER\", 'localhost:9092')\n",
    "\n",
    "SPARK_APP_NAME = \"Final - PSPD - Predict\"\n",
    "INTERVAL = os.getenv(\"INTERVAL\", \"10 seconds\")\n",
    "\n",
    "PREDICT_TOPIC = os.getenv(\"PREDICT_TOPIC\", 'election')\n",
    "STATS_TOPIC = os.getenv(\"STATS_TOPIC\", 'test-elasticsearch-sink')\n",
    "\n",
    "PACKAGES = \"org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0\"\n",
    "\n",
    "PRETRAINED_MODEL_PATH = os.getenv(\"PRETRAINED_MODEL_PATH\", \"model/trained.model\")\n",
    "STOPWORDS_PATH = os.getenv(\"STOPWORDS_PATH\", \"dataset/stopwords.txt\")\n",
    "\n",
    "SPARK_CORES_MAX = os.getenv(\"SPARK_CORES_MAX\", \"2\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a1e16a5f-aacf-47e0-a96a-60bd32f65b01",
   "metadata": {},
   "source": [
    "## Startup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "7ecfb9ea-fbbf-41c3-876f-674535dfd199",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/home/thiago/.local/lib/python3.10/site-packages/pyspark/jars/ivy-2.5.0.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Ivy Default Cache set to: /home/thiago/.ivy2/cache\n",
      "The jars for the packages stored in: /home/thiago/.ivy2/jars\n",
      "org.apache.spark#spark-sql-kafka-0-10_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-911440c7-d108-4004-8b1a-777d05ec1389;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.apache.spark#spark-sql-kafka-0-10_2.12;3.2.0 in central\n",
      "\tfound org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.2.0 in central\n",
      "\tfound org.apache.kafka#kafka-clients;2.8.0 in central\n",
      "\tfound org.lz4#lz4-java;1.7.1 in central\n",
      "\tfound org.xerial.snappy#snappy-java;1.1.8.4 in central\n",
      "\tfound org.slf4j#slf4j-api;1.7.30 in central\n",
      "\tfound org.apache.hadoop#hadoop-client-runtime;3.3.1 in central\n",
      "\tfound org.spark-project.spark#unused;1.0.0 in central\n",
      "\tfound org.apache.hadoop#hadoop-client-api;3.3.1 in central\n",
      "\tfound org.apache.htrace#htrace-core4;4.1.0-incubating in central\n",
      "\tfound commons-logging#commons-logging;1.1.3 in central\n",
      "\tfound com.google.code.findbugs#jsr305;3.0.0 in central\n",
      "\tfound org.apache.commons#commons-pool2;2.6.2 in central\n",
      ":: resolution report :: resolve 717ms :: artifacts dl 41ms\n",
      "\t:: modules in use:\n",
      "\tcom.google.code.findbugs#jsr305;3.0.0 from central in [default]\n",
      "\tcommons-logging#commons-logging;1.1.3 from central in [default]\n",
      "\torg.apache.commons#commons-pool2;2.6.2 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-client-api;3.3.1 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-client-runtime;3.3.1 from central in [default]\n",
      "\torg.apache.htrace#htrace-core4;4.1.0-incubating from central in [default]\n",
      "\torg.apache.kafka#kafka-clients;2.8.0 from central in [default]\n",
      "\torg.apache.spark#spark-sql-kafka-0-10_2.12;3.2.0 from central in [default]\n",
      "\torg.apache.spark#spark-token-provider-kafka-0-10_2.12;3.2.0 from central in [default]\n",
      "\torg.lz4#lz4-java;1.7.1 from central in [default]\n",
      "\torg.slf4j#slf4j-api;1.7.30 from central in [default]\n",
      "\torg.spark-project.spark#unused;1.0.0 from central in [default]\n",
      "\torg.xerial.snappy#snappy-java;1.1.8.4 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   13  |   0   |   0   |   0   ||   13  |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-911440c7-d108-4004-8b1a-777d05ec1389\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 13 already retrieved (0kB/28ms)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/18 22:36:32 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "conf = SparkConf() \\\n",
    "    .setMaster(SPARK_MASTER) \\\n",
    "    .setAppName(SPARK_APP_NAME) \\\n",
    "    .set(\"spark.jars.packages\", PACKAGES) \\\n",
    "    .set(\"spark.cores.max\", \"2\")\n",
    "    \n",
    "context = SparkContext(conf=conf)\n",
    "context.setLogLevel(\"ERROR\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "46c7cda4-1a80-4596-ae4a-eb2709436582",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2bbbfaf5-8bed-4e2b-8944-9a9552096258",
   "metadata": {},
   "source": [
    "## Cleaner"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "d5bd50a9-1d8f-466d-8d6f-f8a569b704be",
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "\n",
    "CLEAN_REGEX = r\"[.,/\\\\\\[\\]\\{\\}`~^\\d&!@#$%*\\)\\(\\'\\\"<>=+-:;?“]\"\n",
    "\n",
    "stopwords = set()\n",
    "\n",
    "with open(STOPWORDS_PATH, \"r\") as stop_file:\n",
    "    for w in stop_file:\n",
    "        stopwords.add(w.strip().lower())\n",
    "\n",
    "def cleaner(sentence):\n",
    "    sentence = \" \".join(\n",
    "        filter(\n",
    "            lambda x: x not in stopwords,\n",
    "            re.sub(CLEAN_REGEX, '', sentence).split()\n",
    "        )\n",
    "    )\n",
    "    return sentence\n",
    "\n",
    "cleaner_col = udf(lambda s: cleaner(s), StringType())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "203786bf",
   "metadata": {},
   "source": [
    "## Load Pre-trained Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "1c56c630",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "model = PipelineModel.load(PRETRAINED_MODEL_PATH)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7f0446fa-8a4d-419d-a6d6-5955ca1c13e3",
   "metadata": {},
   "source": [
    "## Prediction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "6b31a2c1-70c5-42e8-889c-6dff59b78f29",
   "metadata": {},
   "outputs": [],
   "source": [
    "def foreach_batch_func(df: DataFrame, _):\n",
    "    # Preparations - split into candidate and message and clean\n",
    "    candidateMessage = split(df.value, \",\", 2)\n",
    "    sentences = df \\\n",
    "                .withColumn(\"candidate\", candidateMessage.getItem(0)) \\\n",
    "                .withColumn(\"sentence\", cleaner_col(lower(candidateMessage.getItem(1))))\n",
    "\n",
    "    # Predict\n",
    "    prediction = model.transform(sentences) \\\n",
    "                .select(\n",
    "                    \"candidate\",\n",
    "                    \"sentence\",\n",
    "                    \"probability\",\n",
    "                    when(col(\"prediction\") == 1.0, \"positive\").otherwise(\"negative\").alias(\"prediction\")\n",
    "                ) \\\n",
    "\n",
    "    # Write in console\n",
    "    prediction \\\n",
    "        .write \\\n",
    "        .format(\"console\") \\\n",
    "        .save()\n",
    "\n",
    "    # Prepare prediction to elasticsearch format\n",
    "    # Group by candidate and prediction and format to json\n",
    "    predictionElastic = prediction \\\n",
    "                        .groupBy(\n",
    "                            \"candidate\",\n",
    "                            \"prediction\"\n",
    "                        ).count() \\\n",
    "                        .select(\n",
    "                            lit('1').alias(\"key\"),\n",
    "                            format_string(\n",
    "                                \"{\\\"candidate\\\": \\\"%s\\\", \\\"%s\\\": %d}\",\n",
    "                                col(\"candidate\"), col(\"prediction\"), col(\"count\")\n",
    "                            ).alias(\"value\")\n",
    "                        )\n",
    "    \n",
    "    # Write to kafka elasticsearch topic\n",
    "    predictionElastic.write \\\n",
    "                    .format(\"kafka\") \\\n",
    "                    .option(\"kafka.bootstrap.servers\", KAFKA_SERVER) \\\n",
    "                    .option('topic', STATS_TOPIC) \\\n",
    "                    .save()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18ee39d3-be33-4eec-b043-156b1c81a5be",
   "metadata": {},
   "source": [
    "## Sink"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "ec95c249-6eeb-49a5-ae56-53c6910e20d5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+-----------+----------+\n",
      "|candidate|sentence|probability|prediction|\n",
      "+---------+--------+-----------+----------+\n",
      "+---------+--------+-----------+----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------------------+----------+\n",
      "|candidate|            sentence|         probability|prediction|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "|     Lula|rt centraleleicoe...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt silviogrimaldo...|[0.01699080700473...|  positive|\n",
      "|     Lula|voto voto voto co...|[0.05473215335437...|  positive|\n",
      "|     Lula|andrejanonesadv m...|[0.99999846087890...|  negative|\n",
      "|     Lula|rt mary_brunna de...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt nelly pau ferr...|[0.74190901942733...|  negative|\n",
      "|     Lula|rt acheiteresa te...|[0.88172004002475...|  negative|\n",
      "|     Lula|rt monederojc eeu...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt thiagoresiste ...|[0.01699080700473...|  positive|\n",
      "|     Lula|thiagoresiste lul...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt silviogrimaldo...|[0.01699080700473...|  positive|\n",
      "|Bolsonaro|rt dalcolgiovane ...|[0.64888942855328...|  negative|\n",
      "|Bolsonaro|rt filipesabara d...|[0.96631028672029...|  negative|\n",
      "|Bolsonaro|burro achando mor...|[0.96609552385420...|  negative|\n",
      "|Bolsonaro|voto voto voto co...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt mohamad partid...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt xicosa cheques...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt mary_brunna de...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt waltshaub don’...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt carlacristinar...|[0.99694368092482...|  negative|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------------------+----------+\n",
      "|candidate|            sentence|         probability|prediction|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "|     Lula|rt arquivoxandy m...|[0.01022091803771...|  positive|\n",
      "|     Lula|rt pedroronchi tr...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt diocla governa...|[0.99844748193180...|  negative|\n",
      "|     Lula|anaaraphaela nils...|[0.36659381609298...|  positive|\n",
      "|     Lula|rt joaquimmonstra...|[0.24767692088327...|  positive|\n",
      "|     Lula|marelhos cher_gue...|[0.28786085655245...|  positive|\n",
      "|     Lula|rt geraldoalckmin...|[0.37168065933096...|  positive|\n",
      "|     Lula|vestia olhem cami...|[4.59582117257998...|  positive|\n",
      "|     Lula|andreaad monark l...|[0.96631028672029...|  negative|\n",
      "|     Lula|pedroronchi salár...|[0.04013510271677...|  positive|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------------------+----------+\n",
      "|candidate|            sentence|         probability|prediction|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "|Bolsonaro|rt ale_pavanelli ...|[0.98326493015951...|  negative|\n",
      "|Bolsonaro|rt medoedeliriobr...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt benitorperez ⁦...|[0.89542213740066...|  negative|\n",
      "|Bolsonaro|rt jpdoficial ree...|[0.04274809677888...|  positive|\n",
      "|Bolsonaro|lulaverso cabra v...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt thiagoresiste ...|[0.19203629038244...|  positive|\n",
      "|Bolsonaro|rt leandroruschel...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt pedroronchi le...|[0.81417785731452...|  negative|\n",
      "|Bolsonaro|rt lira president...|[0.37423303445666...|  positive|\n",
      "|Bolsonaro|rt carlosjordy ca...|[0.01699080700473...|  positive|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------------------+----------+\n",
      "|candidate|            sentence|         probability|prediction|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "|     Lula|rt kimpaim eleito...|[0.00513322302445...|  positive|\n",
      "|     Lula|rt pedroronchi en...|[0.00398240512232...|  positive|\n",
      "|     Lula|rt kimpaim eleito...|[0.00513322302445...|  positive|\n",
      "|     Lula|jairmearrependi p...|[0.75124716926032...|  negative|\n",
      "|     Lula|rt umdedodearte f...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt eliasjabbour i...|[0.01699080700473...|  positive|\n",
      "|     Lula|rt kimpaim eleito...|[0.00513322302445...|  positive|\n",
      "|     Lula|rt celle_a_celle ...|[0.12031581956437...|  positive|\n",
      "|     Lula|costajr lulaofici...|[0.34420151030417...|  positive|\n",
      "|     Lula|rt josivamalves l...|[0.36659381609298...|  positive|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------------------+--------------------+----------+\n",
      "|   candidate|            sentence|         probability|prediction|\n",
      "+------------+--------------------+--------------------+----------+\n",
      "|   Bolsonaro|rt netinhoespinha...|[0.08049147681987...|  positive|\n",
      "|   Bolsonaro|rt dilmaresiste w...|[0.00943102952331...|  positive|\n",
      "|   Bolsonaro|rt athenasmgf mer...|[0.11383431800661...|  positive|\n",
      "|   Bolsonaro|rt flvialeo belís...|[0.03922312990939...|  positive|\n",
      "|   Bolsonaro|rt rafaelbboa err...|[0.05473215335437...|  positive|\n",
      "|   Bolsonaro|rt senadorhumbert...|[0.71045025666700...|  negative|\n",
      "|   Bolsonaro|rt bernilton paul...|[2.41930989551548...|  positive|\n",
      "|   Bolsonaro|rt pedroronchi en...|[0.00398240512232...|  positive|\n",
      "|   Bolsonaro|rt jamilchade exc...|[0.05473215335437...|  positive|\n",
      "|   Bolsonaro|rt profpaulamaris...|[0.05473215335437...|  positive|\n",
      "|Simone Tebet|rt henriolliveira...|[0.41434642448243...|  positive|\n",
      "|Simone Tebet|rt henriolliveira...|[0.41434642448243...|  positive|\n",
      "|Simone Tebet|sabemos sente mas...|[0.01045323742220...|  positive|\n",
      "|Simone Tebet|rt walfridofreire...|[0.00513322302445...|  positive|\n",
      "|Simone Tebet|please help suppo...|[0.01699080700473...|  positive|\n",
      "|Simone Tebet|rt mariana_ metad...|[0.01569685586824...|  positive|\n",
      "|Simone Tebet|rt falarafha simo...|[0.00513322302445...|  positive|\n",
      "|Simone Tebet|rt sidrevoltz sim...|[0.00321816463924...|  positive|\n",
      "|Simone Tebet|rt terrabrasilnot...|[0.01699080700473...|  positive|\n",
      "|Simone Tebet|rt monicabergamo ...|[0.01185384248803...|  positive|\n",
      "+------------+--------------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------------------+----------+\n",
      "|candidate|            sentence|         probability|prediction|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "|     Lula|rt monark ganhara...|[0.30803668091843...|  positive|\n",
      "|     Lula|rt kimpaim tamanh...|[0.01544559309229...|  positive|\n",
      "|     Lula|rt burlamaquip gl...|[0.05473215335437...|  positive|\n",
      "|     Lula|acabou postar víd...|[0.98866585003161...|  negative|\n",
      "|     Lula|rt folha janones ...|[0.00513322302445...|  positive|\n",
      "|     Lula|rt heldersalomao ...|[0.01699080700473...|  positive|\n",
      "|     Lula|luciolacolares lu...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt uolnoticias da...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt viccommie bent...|[0.05293129498728...|  positive|\n",
      "|     Lula|rt damadeferroofi...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|lcpj centraleleic...|[0.84642205838004...|  negative|\n",
      "|Bolsonaro|monark danilogent...|[0.01441770223105...|  positive|\n",
      "|Bolsonaro|rt augustonpistol...|[0.01699080700473...|  positive|\n",
      "|Bolsonaro|rt lulaoficial lo...|[0.14140162804382...|  positive|\n",
      "|Bolsonaro|rt foiha_sp urgen...|[0.03992738801992...|  positive|\n",
      "|Bolsonaro|rt ivanvalente qu...|[0.01699080700473...|  positive|\n",
      "|Bolsonaro|rt ademaralvessil...|[0.80402378587032...|  negative|\n",
      "|Bolsonaro|rt rosapantin y l...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt revistaoeste p...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt _frota explica...|[0.01699080700473...|  positive|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------------------+----------+\n",
      "|candidate|            sentence|         probability|prediction|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "|     Lula|rt kimpaim tamanh...|[0.01544559309229...|  positive|\n",
      "|     Lula|rt _janoninho bra...|[0.99303356470823...|  negative|\n",
      "|     Lula|rt uylibertarios ...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt brianmtelesur ...|[0.09506243406881...|  positive|\n",
      "|     Lula|q voltar pra gent...|[0.66470525795489...|  negative|\n",
      "|     Lula|rt acheiteresa te...|[0.88172004002475...|  negative|\n",
      "|     Lula|rt kimpaim eleito...|[0.00513322302445...|  positive|\n",
      "|     Lula|rt joaogoulartjoa...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt renato_rovai f...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt burlamaquip gl...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt erikamo impren...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt augustonpistol...|[0.01699080700473...|  positive|\n",
      "|Bolsonaro|glauber_braga ade...|[0.00689842520472...|  positive|\n",
      "|Bolsonaro|rt nikolas_dm gos...|[0.27978725229394...|  positive|\n",
      "|Bolsonaro|rt pedroronchi ah...|[0.22699429102587...|  positive|\n",
      "|Bolsonaro|rt denisps faland...|[0.65275819431067...|  negative|\n",
      "|Bolsonaro|rt truckdriverple...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt centraleleicoe...|[0.13595631677014...|  positive|\n",
      "|Bolsonaro|rt augustonsincer...|[0.03170926165201...|  positive|\n",
      "|Bolsonaro|rt damadanoite 🇧...|[0.01699080700473...|  positive|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------------------+----------+\n",
      "|candidate|            sentence|         probability|prediction|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "|     Lula|rt marcianobrito ...|[4.36863700934153...|  positive|\n",
      "|     Lula|rt elaineg bolson...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt burlamaquip gl...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt paulolo hj rin...|[0.96593855111309...|  negative|\n",
      "|     Lula|rt jornalismojoao...|[0.01651828612116...|  positive|\n",
      "|     Lula|revistaoeste inst...|[0.96631028672029...|  negative|\n",
      "|     Lula|compreensível des...|[0.97687882291981...|  negative|\n",
      "|     Lula|rt centraleleicoe...|[0.05473215335437...|  positive|\n",
      "|     Lula|rt arnoldonunes n...|[0.05371745543959...|  positive|\n",
      "|     Lula|rt monark ganhara...|[0.30803668091843...|  positive|\n",
      "|Bolsonaro|adoro pisa nordes...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt alexandrekunz ...|[0.01784463538045...|  positive|\n",
      "|Bolsonaro|rt laderechadiari...|[0.01252358944790...|  positive|\n",
      "|Bolsonaro|rt marcelopjs rel...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt dalcolgiovane ...|[0.04348997842903...|  positive|\n",
      "|Bolsonaro|rt viniciuscfp el...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt uolnoticias ed...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt freu_rodrigues...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt burlamaquip gl...|[0.05473215335437...|  positive|\n",
      "|Bolsonaro|rt gleisi importa...|[3.39378223288399...|  positive|\n",
      "+---------+--------------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "lines = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", KAFKA_SERVER) \\\n",
    "    .option(\"subscribe\", PREDICT_TOPIC) \\\n",
    "    .option(\"failOnDataLoss\", \"false\") \\\n",
    "    .load() \\\n",
    "    .writeStream \\\n",
    "    .foreachBatch(foreach_batch_func) \\\n",
    "    .option(\"checkpointLocation\", \"/tmp/spark/mllib-predict\") \\\n",
    "    .trigger(processingTime=INTERVAL) \\\n",
    "    .start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "63b62552-5398-48a7-ae97-852f095f7b87",
   "metadata": {},
   "source": [
    "# End"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "bc721391-1f82-4344-bd48-e672569b95fe",
   "metadata": {},
   "outputs": [],
   "source": [
    "lines.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "b0a84f92-467a-4e3a-bf9f-f734226abc50",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()\n",
    "context.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1a6f7091-0304-4793-b4c6-5d8143c250a8",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.7"
  },
  "vscode": {
   "interpreter": {
    "hash": "ed51b0d4d802ee6bae50f5ea08abdf9a7a07ae976ce5019e7e29aaee23a066c8"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}