practice-pyflink/notebook/6.tableapi-malay-sentiment-classifier-kafka.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b71efddd",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyflink.table import DataTypes, TableEnvironment, EnvironmentSettings\n",
    "from pyflink.table.expressions import lit, call\n",
    "from pyflink.table.udf import ScalarFunction\n",
    "from pyflink.table.udf import udf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "86b56175",
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "\n",
    "def cleaning(string):\n",
    "    string = ' '.join([w for w in string.split() if w.find('@') < 0])\n",
    "    string = re.sub('\\(dot\\)', '.', string)\n",
    "    string = (\n",
    "        re.sub(re.findall(r'\\<a(.*?)\\>', string)[0], '', string)\n",
    "        if (len(re.findall(r'\\<a (.*?)\\>', string)) > 0)\n",
    "        and ('href' in re.findall(r'\\<a (.*?)\\>', string)[0])\n",
    "        else string\n",
    "    )\n",
    "    string = re.sub(\n",
    "        r'\\w+:\\/{2}[\\d\\w-]+(\\.[\\d\\w-]+)*(?:(?:\\/[^\\s/]*))*', ' ', string\n",
    "    )\n",
    "    string = re.sub('[^A-Za-z0-9 ]+', ' ', string)\n",
    "    string = re.sub(r'[ ]+', ' ', string).strip()\n",
    "    return string.lower()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "e462ef00",
   "metadata": {},
   "outputs": [],
   "source": [
    "settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()\n",
    "t_env = TableEnvironment.create(settings)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "532a1540",
   "metadata": {},
   "outputs": [],
   "source": [
    "class SentimentClassifier(ScalarFunction):\n",
    "    def __init__(self, filename = 'tfidf-nb-malay-sentiment.pkl'):\n",
    "        import pydoop.hdfs\n",
    "        import os\n",
    "        import pickle\n",
    "        \n",
    "        hdfs = pydoop.hdfs.hdfs(host = 'hdfs', port = 9000)\n",
    "        with hdfs.open_file(os.path.join('/user', filename), 'rb') as fopen:\n",
    "            self.model = pickle.loads(fopen.read())\n",
    "\n",
    "    def eval(self, string):\n",
    "        return str(self.model.predict([cleaning(string)])[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "35b714eb",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-03-04 04:32:04,048 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    }
   ],
   "source": [
    "classifier = SentimentClassifier()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "453b3460",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Negative'"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "classifier.eval('saya busuk')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "7fc1d83e",
   "metadata": {},
   "outputs": [],
   "source": [
    "sentiment_classifier = udf(classifier, result_type=DataTypes.STRING())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "3ee503d5",
   "metadata": {},
   "outputs": [],
   "source": [
    "t_env.register_function('sentiment_classifier', sentiment_classifier)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "e18abfb3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# !wget https://raw.githubusercontent.com/huseinzol05/malay-dataset/master/sentiment/supervised-twitter/data.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "ec61564e",
   "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>text</th>\n",
       "      <th>id</th>\n",
       "      <th>sentiment</th>\n",
       "      <th>annotator</th>\n",
       "      <th>annotation_id</th>\n",
       "      <th>created_at</th>\n",
       "      <th>updated_at</th>\n",
       "      <th>lead_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>perempuan tu apesal aku rasa mcm pernah nampak...</td>\n",
       "      <td>68381</td>\n",
       "      <td>Neutral</td>\n",
       "      <td>27</td>\n",
       "      <td>4603</td>\n",
       "      <td>2022-02-15T14:04:28.798425Z</td>\n",
       "      <td>2022-02-15T14:04:32.792755Z</td>\n",
       "      <td>21974.040</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Polis tangkap dia.</td>\n",
       "      <td>68345</td>\n",
       "      <td>Negative</td>\n",
       "      <td>27</td>\n",
       "      <td>4567</td>\n",
       "      <td>2022-02-15T12:37:58.202450Z</td>\n",
       "      <td>2022-02-15T12:38:00.815718Z</td>\n",
       "      <td>16782.079</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Kenapa lokasi kebakaran nya terlalu spesifik? ...</td>\n",
       "      <td>68325</td>\n",
       "      <td>Negative</td>\n",
       "      <td>27</td>\n",
       "      <td>4547</td>\n",
       "      <td>2022-02-15T11:47:23.902859Z</td>\n",
       "      <td>2022-02-15T11:47:27.849352Z</td>\n",
       "      <td>13749.111</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>@YuleumSummer Waa senang nye lah nk bersalin.....</td>\n",
       "      <td>68286</td>\n",
       "      <td>Positive</td>\n",
       "      <td>27</td>\n",
       "      <td>4508</td>\n",
       "      <td>2022-02-15T08:59:24.606503Z</td>\n",
       "      <td>2022-02-15T08:59:27.475987Z</td>\n",
       "      <td>3668.697</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>DAULAT TUANKU! Merafak sembah dan takzim KDYMM...</td>\n",
       "      <td>68247</td>\n",
       "      <td>Positive</td>\n",
       "      <td>27</td>\n",
       "      <td>4469</td>\n",
       "      <td>2022-02-15T08:34:10.363218Z</td>\n",
       "      <td>2022-02-15T08:34:13.342956Z</td>\n",
       "      <td>2154.577</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                text     id sentiment  \\\n",
       "0  perempuan tu apesal aku rasa mcm pernah nampak...  68381   Neutral   \n",
       "1                                 Polis tangkap dia.  68345  Negative   \n",
       "2  Kenapa lokasi kebakaran nya terlalu spesifik? ...  68325  Negative   \n",
       "3  @YuleumSummer Waa senang nye lah nk bersalin.....  68286  Positive   \n",
       "4  DAULAT TUANKU! Merafak sembah dan takzim KDYMM...  68247  Positive   \n",
       "\n",
       "   annotator  annotation_id                   created_at  \\\n",
       "0         27           4603  2022-02-15T14:04:28.798425Z   \n",
       "1         27           4567  2022-02-15T12:37:58.202450Z   \n",
       "2         27           4547  2022-02-15T11:47:23.902859Z   \n",
       "3         27           4508  2022-02-15T08:59:24.606503Z   \n",
       "4         27           4469  2022-02-15T08:34:10.363218Z   \n",
       "\n",
       "                    updated_at  lead_time  \n",
       "0  2022-02-15T14:04:32.792755Z  21974.040  \n",
       "1  2022-02-15T12:38:00.815718Z  16782.079  \n",
       "2  2022-02-15T11:47:27.849352Z  13749.111  \n",
       "3  2022-02-15T08:59:27.475987Z   3668.697  \n",
       "4  2022-02-15T08:34:13.342956Z   2154.577  "
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pandas as pd\n",
    "\n",
    "filename = 'data.csv'\n",
    "df = pd.read_csv(filename, sep = '\\t')\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "e31fc345",
   "metadata": {},
   "outputs": [],
   "source": [
    "columns = df.columns.tolist()\n",
    "columns = ', '.join([f'`{c}` VARCHAR' for c in columns])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "11bf1010",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pydoop.hdfs\n",
    "import os\n",
    "\n",
    "hdfs = pydoop.hdfs.hdfs(host = 'hdfs', port = 9000)\n",
    "\n",
    "with hdfs.open_file(os.path.join('/user', filename), 'wb') as fopen:\n",
    "    with open(filename, 'r') as fopen_csv:\n",
    "        fopen.write(fopen_csv.read().encode())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "f77c26b8",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyflink.table.table_result.TableResult at 0x7f005d753c70>"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "hdfs_path_input = f'hdfs://hdfs:9000/user/{filename}'\n",
    "\n",
    "my_source_ddl = f\"\"\"\n",
    "    create table mySource (\n",
    "        {columns}\n",
    "    ) with (\n",
    "        'connector' = 'filesystem',\n",
    "        'format' = 'csv',\n",
    "        'csv.field-delimiter' = '\\t',\n",
    "        'path' = '{hdfs_path_input}'\n",
    "    )\n",
    "\"\"\"\n",
    "\n",
    "my_sink_ddl = f\"\"\"\n",
    "    create table mySink (\n",
    "        sentence VARCHAR,\n",
    "        label VARCHAR\n",
    "    ) with (\n",
    "        'connector' = 'kafka',\n",
    "        'topic' = 'employee.public.employee\n",
    "',\n",
    "        'properties.bootstrap.servers' = 'broker:9092',\n",
    "        'properties.group.id' = 'testGroup',\n",
    "        'scan.startup.mode' = 'earliest-offset',\n",
    "        'format' = 'json'\n",
    "    )\n",
    "\"\"\"\n",
    "\n",
    "t_env.execute_sql(my_source_ddl)\n",
    "t_env.execute_sql(my_sink_ddl)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "2efa91be",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyflink.table.table.Table at 0x7f00a4251f40>"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tab = t_env.from_path('mySource').offset(1)\n",
    "tab"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "b6c427da",
   "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>text</th>\n",
       "      <th>id</th>\n",
       "      <th>sentiment</th>\n",
       "      <th>annotator</th>\n",
       "      <th>annotation_id</th>\n",
       "      <th>created_at</th>\n",
       "      <th>updated_at</th>\n",
       "      <th>lead_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>perempuan tu apesal aku rasa mcm pernah nampak...</td>\n",
       "      <td>68381</td>\n",
       "      <td>Neutral</td>\n",
       "      <td>27</td>\n",
       "      <td>4603</td>\n",
       "      <td>2022-02-15T14:04:28.798425Z</td>\n",
       "      <td>2022-02-15T14:04:32.792755Z</td>\n",
       "      <td>21974.04</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Polis tangkap dia.</td>\n",
       "      <td>68345</td>\n",
       "      <td>Negative</td>\n",
       "      <td>27</td>\n",
       "      <td>4567</td>\n",
       "      <td>2022-02-15T12:37:58.202450Z</td>\n",
       "      <td>2022-02-15T12:38:00.815718Z</td>\n",
       "      <td>16782.079</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Kenapa lokasi kebakaran nya terlalu spesifik? ...</td>\n",
       "      <td>68325</td>\n",
       "      <td>Negative</td>\n",
       "      <td>27</td>\n",
       "      <td>4547</td>\n",
       "      <td>2022-02-15T11:47:23.902859Z</td>\n",
       "      <td>2022-02-15T11:47:27.849352Z</td>\n",
       "      <td>13749.111</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>@YuleumSummer Waa senang nye lah nk bersalin.....</td>\n",
       "      <td>68286</td>\n",
       "      <td>Positive</td>\n",
       "      <td>27</td>\n",
       "      <td>4508</td>\n",
       "      <td>2022-02-15T08:59:24.606503Z</td>\n",
       "      <td>2022-02-15T08:59:27.475987Z</td>\n",
       "      <td>3668.697</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>DAULAT TUANKU! Merafak sembah dan takzim KDYMM...</td>\n",
       "      <td>68247</td>\n",
       "      <td>Positive</td>\n",
       "      <td>27</td>\n",
       "      <td>4469</td>\n",
       "      <td>2022-02-15T08:34:10.363218Z</td>\n",
       "      <td>2022-02-15T08:34:13.342956Z</td>\n",
       "      <td>2154.577</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                text     id sentiment  \\\n",
       "0  perempuan tu apesal aku rasa mcm pernah nampak...  68381   Neutral   \n",
       "1                                 Polis tangkap dia.  68345  Negative   \n",
       "2  Kenapa lokasi kebakaran nya terlalu spesifik? ...  68325  Negative   \n",
       "3  @YuleumSummer Waa senang nye lah nk bersalin.....  68286  Positive   \n",
       "4  DAULAT TUANKU! Merafak sembah dan takzim KDYMM...  68247  Positive   \n",
       "\n",
       "  annotator annotation_id                   created_at  \\\n",
       "0        27          4603  2022-02-15T14:04:28.798425Z   \n",
       "1        27          4567  2022-02-15T12:37:58.202450Z   \n",
       "2        27          4547  2022-02-15T11:47:23.902859Z   \n",
       "3        27          4508  2022-02-15T08:59:24.606503Z   \n",
       "4        27          4469  2022-02-15T08:34:10.363218Z   \n",
       "\n",
       "                    updated_at  lead_time  \n",
       "0  2022-02-15T14:04:32.792755Z   21974.04  \n",
       "1  2022-02-15T12:38:00.815718Z  16782.079  \n",
       "2  2022-02-15T11:47:27.849352Z  13749.111  \n",
       "3  2022-02-15T08:59:27.475987Z   3668.697  \n",
       "4  2022-02-15T08:34:13.342956Z   2154.577  "
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tab.to_pandas().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "b97ba8d5",
   "metadata": {},
   "outputs": [],
   "source": [
    "tab.select('text, sentiment_classifier(text) as label') \\\n",
    "   .execute_insert('mySink').wait()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "f2b9a0b6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# tab = t_env.from_path('mySink')\n",
    "# tab.to_pandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "5b7a6e99",
   "metadata": {},
   "outputs": [],
   "source": [
    "# !pip3 install confluent-kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "d0fe5a38",
   "metadata": {},
   "outputs": [],
   "source": [
    "from confluent_kafka import Consumer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "36c0a5fe",
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = {'bootstrap.servers': 'broker:9092', 'group.id': 'test', 'session.timeout.ms': 6000,\n",
    "            'auto.offset.reset': 'earliest'}\n",
    "c = Consumer(conf)\n",
    "c.subscribe(['mysink_sentiment'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "97fbec39",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<cimpl.Message at 0x7f002a0aad40>"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "message = c.poll(0)\n",
    "message"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "450557a2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "b'{\"sentence\":\"perempuan tu apesal aku rasa mcm pernah nampak?? student unisel ke cane eh\",\"label\":\"Negative\"}'"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "message.value()"
   ]
  }
 ],
 "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.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}