spark.ipynb

{
 "cells": [
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "\n",
    "findspark.init()\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark\n",
    "\"\"\" conf = pyspark.SparkConf().setAppName('Tap').setMaster('local[4]')\n",
    "conf.set(\"spark.network.timeout\", \"360000s\")\n",
    "conf.set(\"spark.executor.heartbeatInterval\", \"36000s\")\n",
    "sc = SparkContext.getOrCreate(conf=conf)\n",
    "spark = SparkSession(sc) \"\"\"\n",
    "spark = (\n",
    "    # https://luminousmen.com/post/how-to-speed-up-spark-jobs-on-small-test-datasets?trk=article-ssr-frontend-pulse_little-text-block\n",
    "    SparkSession.builder.appName(\"Spark Jupyter\")\n",
    "    .master(\"local[1]\")\n",
    "    .config(\"spark.driver.memory\", \"24G\")\n",
    "    .config(\"spark.driver.maxResultSize\", \"0\")\n",
    "    .config(\"spark.serializer\", \"org.apache.spark.serializer.KryoSerializer\")\n",
    "    # .config(\"spark.kryoserializer.buffer.max\", \"2000M\")\n",
    "    .config(\"spark.jars.packages\", \"com.johnsnowlabs.nlp:spark-nlp-gpu_2.12:5.4.0\")\n",
    "    .config(\"spark.local.dir\", \"C:/tmp\")\n",
    "    # https://stackoverflow.com/questions/34625410/why-does-my-spark-run-slower-than-pure-python-performance-comparison\n",
    "    .config(\"spark.sql.shuffle.partitions\", \"1\")\n",
    "    .config(\"spark.default.parallelism\", \"1\")\n",
    "    .config(\"spark.rdd.compress\", \"false\")\n",
    "    .config(\"spark.shuffle.compress\", \"false\")\n",
    "    .getOrCreate()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://host.docker.internal:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.5.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[1]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Spark Jupyter</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x14d3ed1ad90>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "DATASET_PATH = \"./dataset/dataset.csv\"\n",
    "IMAGES_PATH = \"./dataset/images/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.types as tp\n",
    "#  id,title,subreddit,score,num_comments,created_utc,author,upvote_ratio,img_filename,ocr_text,url,selftext,img_url\n",
    "posts_schema = tp.StructType(\n",
    "    [\n",
    "        tp.StructField(name=\"id\", dataType=tp.StringType(), nullable=False),\n",
    "        tp.StructField(name=\"title\", dataType=tp.StringType(), nullable=False),\n",
    "        tp.StructField(name=\"subreddit\", dataType=tp.StringType(), nullable=False),\n",
    "        tp.StructField(name=\"score\", dataType=tp.IntegerType(), nullable=False),\n",
    "        tp.StructField(name=\"num_comments\", dataType=tp.IntegerType(), nullable=False),\n",
    "        tp.StructField(name=\"created_utc\", dataType=tp.FloatType(), nullable=True),\n",
    "        tp.StructField(name=\"author\", dataType=tp.StringType(), nullable=False),\n",
    "        tp.StructField(name=\"upvote_ratio\", dataType=tp.FloatType(), nullable=False),\n",
    "        tp.StructField(name=\"img_filename\", dataType=tp.StringType(), nullable=False),\n",
    "        tp.StructField(name=\"ocr_text\", dataType=tp.StringType(), nullable=False),\n",
    "        tp.StructField(name=\"url\", dataType=tp.StringType(), nullable=False),\n",
    "        tp.StructField(name=\"selftext\", dataType=tp.StringType(), nullable=True),\n",
    "        tp.StructField(name=\"img_url\", dataType=tp.StringType(), nullable=True),\n",
    "        tp.StructField(name=\"caption_text\", dataType=tp.StringType(), nullable=True),\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+-------------+-----+------------+-------------------+--------------------+------------+------------+--------------------+--------------------+--------+--------------------+--------------------+\n",
      "|     id|               title|    subreddit|score|num_comments|        created_utc|              author|upvote_ratio|img_filename|            ocr_text|                 url|selftext|             img_url|        caption_text|\n",
      "+-------+--------------------+-------------+-----+------------+-------------------+--------------------+------------+------------+--------------------+--------------------+--------+--------------------+--------------------+\n",
      "|1fj1pz3|Upper body after ...|     GymMemes|   24|           7|2024-09-17 17:10:56|       Schitts-Creek|        0.84|1fj1pz3.jpeg|                   G|https://i.redd.it...|        |./dataset/images/...|a sponge sponge i...|\n",
      "|1fiya02|The Champions Lea...|  soccermemes|   77|           4|2024-09-17 14:45:52|          leyladexxx|        0.99|1fiya02.jpeg|                NULL|https://i.redd.it...|        |./dataset/images/...|a close up of a w...|\n",
      "|1fiti54|Bro just made som...|     GymMemes|  446|          16|2024-09-17 10:12:48|     BlueDragonClipz|        0.98|1fiti54.jpeg|Friend: \"How do I...|https://i.redd.it...|        |./dataset/images/...|a picture of a ma...|\n",
      "|1fit0x1|My squats go deep...|     GymMemes|  223|          23|2024-09-17 09:38:40|             ntkwwwm|        0.97|1fit0x1.jpeg|Car salesman: *sl...|https://i.redd.it...|        |./dataset/images/...|a pair of shoes w...|\n",
      "|1figoik|  Lord have mercy...|footballmemes|   27|           6|2024-09-16 23:32:48|          casualfan2|        0.91|1figoik.jpeg|9 Breaking: Dani ...|https://i.redd.it...|        |./dataset/images/...|an image of a man...|\n",
      "|1fi959m|        Penal Padrid|  soccermemes|   16|           2|2024-09-16 18:32:00|          leyladexxx|        0.81|1fi959m.jpeg|4 penalties in 5 ...|https://i.redd.it...|        |./dataset/images/...|a mememe with fou...|\n",
      "|1fi742v|oh spuds my belov...|  soccermemes|    3|           2|2024-09-16 17:13:04|         sh08ismaili|         0.6|1fi742v.jpeg|   TOTTENHAM HOTSPUR|https://i.redd.it...|        |./dataset/images/...|toteman hotspir -...|\n",
      "|1fi5cx8|Every time. Now I...|     GymMemes|  674|           9|2024-09-16 16:00:32|       QueasyVisuals|        0.99|1fi5cx8.jpeg|What my grandpare...|https://i.redd.it...|        |./dataset/images/...|what my grandpare...|\n",
      "|1fi2hse|Man City’s legal ...|  soccermemes| 1113|          23|2024-09-16 13:46:08|          merindina1|        0.91|1fi2hse.jpeg|ETIAAD STADIUM ET...|https://i.redd.it...|        |./dataset/images/...|a group of people...|\n",
      "|1fi2bba|Champions League ...|  soccermemes|    0|           2|2024-09-16 13:37:36|          leyladexxx|         0.4|1fi2bba.jpeg|                NULL|https://i.redd.it...|        |./dataset/images/...|a soccer ball on ...|\n",
      "|1fhr909|What leaving Chel...|  soccermemes|  102|           3|2024-09-16 02:10:40|          leyladexxx|        0.93|1fhr909.jpeg|    74 9.0 11 Lukaku|https://i.redd.it...|        |./dataset/images/...|a black t shirt w...|\n",
      "|1fhot0d|Ney watching the ...|  soccermemes|    7|           1|2024-09-16 00:15:28|      inactiveandead|        0.89|1fhot0d.jpeg|         neymarjr 5m|https://i.redd.it...|        |./dataset/images/...|a man with a gree...|\n",
      "|1fhmfgc|I hope this isn't...|     GymMemes|  704|          38|2024-09-15 22:33:04|shitposting-gymmemes|        0.98| 1fhmfgc.png|How my mom thinks...|https://i.redd.it...|        |./dataset/images/...|a cartoon drawing...|\n",
      "|1fhhzvi|I'm using that......|     GymMemes|   57|           4|2024-09-15 19:27:28|        outdoor_hawk|        0.93|1fhhzvi.jpeg|Are these [ockers...|https://i.redd.it...|        |./dataset/images/...|are these korean ...|\n",
      "|1fhh5wf| North London is red|  soccermemes|   11|           1|2024-09-15 18:51:12|          leyladexxx|        0.77|1fhh5wf.jpeg|                NULL|https://i.redd.it...|        |./dataset/images/...|a logo with a bir...|\n",
      "|1fhh16w|Arsenal fans be l...|  soccermemes|  188|          28|2024-09-15 18:46:56|   Relevant-Kale6199|        0.76|1fhh16w.jpeg|{Ricegetszndiello...|https://i.redd.it...|        |./dataset/images/...|two pictures of a...|\n",
      "|1fhgei9|Having to make th...|     GymMemes| 1045|          47|2024-09-15 18:19:12|     Lord-Albeit-Fai|        0.98|1fhgei9.jpeg|Gaining weight an...|https://i.redd.it...|        |./dataset/images/...|a man laying in b...|\n",
      "|1fhg8qf| Mbappe and his idol|  soccermemes|   45|           0|2024-09-15 18:12:48|      inactiveandead|        0.95|1fhg8qf.jpeg|                  I6|https://i.redd.it...|        |./dataset/images/...|a young boy sitti...|\n",
      "|1fhfxvm|“Tottenham get ba...|  soccermemes| 1235|          22|2024-09-15 18:00:00|     The_Chuckness88|        0.93| 1fhfxvm.png|THINCS THAT CET B...|https://i.redd.it...|        |./dataset/images/...|a col of differen...|\n",
      "|1fhfml0| Spurs in a NLD w...|  soccermemes|  161|           2|2024-09-15 17:47:12|   Superb-Wonder1406|        0.99| 1fhfml0.png|                Firi|https://i.redd.it...|        |./dataset/images/...|a group of pigeon...|\n",
      "+-------+--------------------+-------------+-----+------------+-------------------+--------------------+------------+------------+--------------------+--------------------+--------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# https://spark.apache.org/docs/latest/sql-data-sources-csv.html\n",
    "# read and impose a schema\n",
    "# df= spark.read.csv(DATASET_PATH, header=True, schema= posts_schema)\n",
    "from pyspark.sql.functions import col, from_unixtime\n",
    "\n",
    "\n",
    "df = (\n",
    "    spark.read.format(\"csv\")\n",
    "    .schema(posts_schema)\n",
    " #   .options(multiline=True, mode=\"FAILFAST\", emptyValue=\"\", nullValue=\"\")\n",
    "    .option(\"multiline\", True)\n",
    "    .option(\"mode\", \"FAILFAST\")\n",
    "    .option(\"emptyValue\", \"\")\n",
    "    .option(\"nullValue\", \"\")\n",
    "    .option(\"delimiter\", \",\")\n",
    "    .option(\"quote\", '\"')\n",
    "    .option(\"escape\", '\"')\n",
    "    .load(DATASET_PATH, header=True)\n",
    ")\n",
    "\n",
    "df = df.withColumn(\n",
    "    \"created_utc\", from_unixtime(col(\"created_utc\")).cast(\"timestamp\")\n",
    ") \n",
    "\n",
    "df = df.fillna({'selftext': ''})\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "540"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# drop if any null column\n",
    "df=df.dropna()\n",
    "\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- subreddit: string (nullable = true)\n",
      " |-- score: integer (nullable = true)\n",
      " |-- num_comments: integer (nullable = true)\n",
      " |-- created_utc: timestamp (nullable = true)\n",
      " |-- author: string (nullable = true)\n",
      " |-- upvote_ratio: float (nullable = true)\n",
      " |-- img_filename: string (nullable = true)\n",
      " |-- ocr_text: string (nullable = true)\n",
      " |-- url: string (nullable = true)\n",
      " |-- selftext: string (nullable = false)\n",
      " |-- img_url: string (nullable = true)\n",
      " |-- caption_text: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+-------------+-----+------------+-------------------+--------------------+------------+------------+--------------------+--------------------+--------+--------------------+--------------------+-----+--------------------+\n",
      "|     id|               title|    subreddit|score|num_comments|        created_utc|              author|upvote_ratio|img_filename|            ocr_text|                 url|selftext|             img_url|        caption_text|label|            all_text|\n",
      "+-------+--------------------+-------------+-----+------------+-------------------+--------------------+------------+------------+--------------------+--------------------+--------+--------------------+--------------------+-----+--------------------+\n",
      "|1fj1pz3|Upper body after ...|     GymMemes|   24|           7|2024-09-17 17:10:56|       Schitts-Creek|        0.84|1fj1pz3.jpeg|                   G|https://i.redd.it...|        |./dataset/images/...|a sponge sponge i...|    0|Upper body after ...|\n",
      "|1fiti54|Bro just made som...|     GymMemes|  446|          16|2024-09-17 10:12:48|     BlueDragonClipz|        0.98|1fiti54.jpeg|Friend: \"How do I...|https://i.redd.it...|        |./dataset/images/...|a picture of a ma...|    0|Bro just made som...|\n",
      "|1fit0x1|My squats go deep...|     GymMemes|  223|          23|2024-09-17 09:38:40|             ntkwwwm|        0.97|1fit0x1.jpeg|Car salesman: *sl...|https://i.redd.it...|        |./dataset/images/...|a pair of shoes w...|    0|My squats go deep...|\n",
      "|1figoik|  Lord have mercy...|footballmemes|   27|           6|2024-09-16 23:32:48|          casualfan2|        0.91|1figoik.jpeg|9 Breaking: Dani ...|https://i.redd.it...|        |./dataset/images/...|an image of a man...|    0|Lord have mercy.....|\n",
      "|1fi959m|        Penal Padrid|  soccermemes|   16|           2|2024-09-16 18:32:00|          leyladexxx|        0.81|1fi959m.jpeg|4 penalties in 5 ...|https://i.redd.it...|        |./dataset/images/...|a mememe with fou...|    0|Penal Padrid  4 p...|\n",
      "|1fi742v|oh spuds my belov...|  soccermemes|    3|           2|2024-09-16 17:13:04|         sh08ismaili|         0.6|1fi742v.jpeg|   TOTTENHAM HOTSPUR|https://i.redd.it...|        |./dataset/images/...|toteman hotspir -...|    0|oh spuds my belov...|\n",
      "|1fi5cx8|Every time. Now I...|     GymMemes|  674|           9|2024-09-16 16:00:32|       QueasyVisuals|        0.99|1fi5cx8.jpeg|What my grandpare...|https://i.redd.it...|        |./dataset/images/...|what my grandpare...|    0|Every time. Now I...|\n",
      "|1fi2hse|Man City’s legal ...|  soccermemes| 1113|          23|2024-09-16 13:46:08|          merindina1|        0.91|1fi2hse.jpeg|ETIAAD STADIUM ET...|https://i.redd.it...|        |./dataset/images/...|a group of people...|    0|Man City’s legal ...|\n",
      "|1fhr909|What leaving Chel...|  soccermemes|  102|           3|2024-09-16 02:10:40|          leyladexxx|        0.93|1fhr909.jpeg|    74 9.0 11 Lukaku|https://i.redd.it...|        |./dataset/images/...|a black t shirt w...|    0|What leaving Chel...|\n",
      "|1fhot0d|Ney watching the ...|  soccermemes|    7|           1|2024-09-16 00:15:28|      inactiveandead|        0.89|1fhot0d.jpeg|         neymarjr 5m|https://i.redd.it...|        |./dataset/images/...|a man with a gree...|    0|Ney watching the ...|\n",
      "|1fhmfgc|I hope this isn't...|     GymMemes|  704|          38|2024-09-15 22:33:04|shitposting-gymmemes|        0.98| 1fhmfgc.png|How my mom thinks...|https://i.redd.it...|        |./dataset/images/...|a cartoon drawing...|    0|I hope this isn't...|\n",
      "|1fhhzvi|I'm using that......|     GymMemes|   57|           4|2024-09-15 19:27:28|        outdoor_hawk|        0.93|1fhhzvi.jpeg|Are these [ockers...|https://i.redd.it...|        |./dataset/images/...|are these korean ...|    0|I'm using that......|\n",
      "|1fhh16w|Arsenal fans be l...|  soccermemes|  188|          28|2024-09-15 18:46:56|   Relevant-Kale6199|        0.76|1fhh16w.jpeg|{Ricegetszndiello...|https://i.redd.it...|        |./dataset/images/...|two pictures of a...|    0|Arsenal fans be l...|\n",
      "|1fhgei9|Having to make th...|     GymMemes| 1045|          47|2024-09-15 18:19:12|     Lord-Albeit-Fai|        0.98|1fhgei9.jpeg|Gaining weight an...|https://i.redd.it...|        |./dataset/images/...|a man laying in b...|    0|Having to make th...|\n",
      "|1fhg8qf| Mbappe and his idol|  soccermemes|   45|           0|2024-09-15 18:12:48|      inactiveandead|        0.95|1fhg8qf.jpeg|                  I6|https://i.redd.it...|        |./dataset/images/...|a young boy sitti...|    0|Mbappe and his id...|\n",
      "|1fhfxvm|“Tottenham get ba...|  soccermemes| 1235|          22|2024-09-15 18:00:00|     The_Chuckness88|        0.93| 1fhfxvm.png|THINCS THAT CET B...|https://i.redd.it...|        |./dataset/images/...|a col of differen...|    0|“Tottenham get ba...|\n",
      "|1fhfml0| Spurs in a NLD w...|  soccermemes|  161|           2|2024-09-15 17:47:12|   Superb-Wonder1406|        0.99| 1fhfml0.png|                Firi|https://i.redd.it...|        |./dataset/images/...|a group of pigeon...|    0| Spurs in a NLD w...|\n",
      "|1fhfcre|        Hell yeah 🗿|     GymMemes| 2325|          48|2024-09-15 17:36:32|          MrDevilxxx|        0.95| 1fhfcre.png|cHowdidyousurvive...|https://i.redd.it...|        |./dataset/images/...|a man and a woman...|    0|Hell yeah 🗿  cHo...|\n",
      "|1fhey6j|Ben Brereton Diaz...|  soccermemes|   16|           0|2024-09-15 17:19:28|  Downtown-Radio6177|        0.88| 1fhey6j.png|B B C SPORT DLLDD...|https://i.redd.it...|        |./dataset/images/...|a poster of a man...|    0|Ben Brereton Diaz...|\n",
      "|1fhe6gc|Why is it like this?|     GymMemes|  350|          15|2024-09-15 16:45:20|           Woahbikes|        0.99|1fhe6gc.jpeg|0 Saaodugerso The...|https://i.redd.it...|        |./dataset/images/...|two men standing ...|    0|Why is it like th...|\n",
      "+-------+--------------------+-------------+-----+------------+-------------------+--------------------+------------+------------+--------------------+--------------------+--------+--------------------+--------------------+-----+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import utils.categories as categories\n",
    "# mappa ogni subreddit alla sua categoria utilizzando il dizionario categories.subreddit_categories\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit\n",
    "\n",
    "\"\"\" def map_subreddit_to_category(subreddit):\n",
    "    return categories.subreddit_categories.get(subreddit, 3)\n",
    "\n",
    "map_subreddit_to_category_udf = udf(map_subreddit_to_category, StringType()) \"\"\"\n",
    "df = df.withColumn(\"label\", udf(lambda subreddit: categories.subreddit_categories.get(subreddit, 3), tp.IntegerType())(col(\"subreddit\")))\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "%%cmd\n",
    "java -version"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "import sparknlp\n",
    "#sparknlp.start()\n",
    "sparknlp.version()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "%%cmd\n",
    "mkdir c:\\tmp\n",
    "mkdir c:\\tmp\\hive\n"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "%%cmd\n",
    "\"%HADOOP_HOME%\\bin\\winutils.exe\" chmod 777 /tmp/hive\n",
    "\"%HADOOP_HOME%\\bin\\winutils.exe\" chmod 777 /tmp/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://stackoverflow.com/questions/65054072/spark-nlp-pretrained-model-not-loading-in-windows\n",
    "https://github.com/JohnSnowLabs/spark-nlp/discussions/1022"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "from sparknlp.base import *\n",
    "from sparknlp.annotator import *\n",
    "\n",
    "from pyspark.ml import Pipeline\n",
    "import os\n",
    "\n",
    "\n",
    "imageDF = spark.read \\\n",
    "    .format(\"image\") \\\n",
    "    .option(\"dropInvalid\", value = True) \\\n",
    "    .load(IMAGES_PATH)\n",
    "imageAssembler = ImageAssembler() \\\n",
    "    .setInputCol(\"image\") \\\n",
    "    .setOutputCol(\"image_assembler\",)\n",
    "imageCaptioning = VisionEncoderDecoderForImageCaptioning \\\n",
    "    .pretrained() \\\n",
    "    .setBeamSize(2) \\\n",
    "    .setDoSample(False) \\\n",
    "    .setInputCols([\"image_assembler\"]) \\\n",
    "    .setOutputCol(\"caption\")\n",
    "\n",
    "pipeline = Pipeline().setStages([imageAssembler, imageCaptioning])\n",
    "pipelineDF = pipeline.fit(imageDF).transform(imageDF)\n",
    "pipelineDF \\\n",
    "    .selectExpr(\"reverse(split(image.origin, '/'))[0] as image_name\", \"caption.result\") \\\n",
    "    .show(truncate = False) "
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "from sparknlp.base import *\n",
    "from sparknlp.annotator import *\n",
    "from pyspark.ml import Pipeline\n",
    "import os\n",
    "from pathlib import Path\n",
    "\n",
    "#CAPTIONING_MODEL_PATH = os.path.abspath(\"./spark/pretrained/captioning/vision_encoder_decoder_tensorflow\")\n",
    "CAPTIONING_MODEL_PATH = Path(\"d:/spark-models/captioning/\").as_uri()\n",
    "IMAGES_PATH = \"./dataset/images/\"\n",
    "\n",
    "\"\"\" # Verifica se il percorso del modello esiste\n",
    "if not os.path.exists(CAPTIONING_MODEL_PATH):\n",
    "    raise FileNotFoundError(f\"Il percorso del modello {CAPTIONING_MODEL_PATH} non esiste.\")\n",
    "else:\n",
    "    print(f\"Il percorso del modello {CAPTIONING_MODEL_PATH} esiste.\")\n",
    "\n",
    "# Verifica i permessi del file\n",
    "if not os.access(CAPTIONING_MODEL_PATH, os.R_OK):\n",
    "    raise PermissionError(f\"Non hai i permessi di lettura per il percorso del modello {CAPTIONING_MODEL_PATH}.\")\n",
    "else:\n",
    "    print(f\"Hai i permessi di lettura per il percorso del modello {CAPTIONING_MODEL_PATH}.\")\n",
    "\n",
    "# Verifica il contenuto del percorso del modello\n",
    "for root, dirs, files in os.walk(CAPTIONING_MODEL_PATH):\n",
    "    for file in files:\n",
    "        print(os.path.join(root, file)) \"\"\"\n",
    "# Carica le immagini\n",
    "imageDF = spark.read \\\n",
    "    .format(\"image\") \\\n",
    "    .option(\"dropInvalid\", value=True) \\\n",
    "    .load(IMAGES_PATH)\n",
    "\n",
    "imageAssembler = ImageAssembler() \\\n",
    "    .setInputCol(\"image\") \\\n",
    "    .setOutputCol(\"image_assembler\")\n",
    "\n",
    "# Carica il modello di captioning\n",
    "try:\n",
    "    imageCaptioning = VisionEncoderDecoderForImageCaptioning \\\n",
    "        .load(CAPTIONING_MODEL_PATH) \\\n",
    "        .setBeamSize(2) \\\n",
    "        .setDoSample(False) \\\n",
    "        .setInputCols([\"image_assembler\"]) \\\n",
    "        .setOutputCol(\"caption\")\n",
    "    print(\"Modello caricato con successo.\")\n",
    "except Exception as e:\n",
    "    print(f\"Errore durante il caricamento del modello: {e}\")\n",
    "    raise\n",
    "\n",
    "pipeline = Pipeline().setStages([imageAssembler, imageCaptioning])\n",
    "pipelineDF = pipeline.fit(imageDF).transform(imageDF)\n",
    "pipelineDF \\\n",
    "    .selectExpr(\"reverse(split(image.origin, '/'))[0] as image_name\", \"caption.result\") \\\n",
    "    .show(truncate=False)"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "pipelineDF.printSchema()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "from pyspark.ml.feature import HashingTF, IDF, Tokenizer\n",
    "# caption.result è un array di stringhe, uniamo le stringhe in un'unica stringa\n",
    "\n",
    "\n",
    "cdf = pipelineDF.withColumn(\"caption_text\", concat_ws(\" \", \"caption.result\"))\n",
    "cdf = cdf.selectExpr(\"reverse(split(image.origin, '/'))[0] as image_name\", \"caption_text\")\n",
    "cdf.show()\n"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "df = df.join(cdf, df.img_filename == cdf.image_name, how=\"inner\")\n",
    "df=df.withColumn(\"label\", df[\"label\"].cast(tp.DoubleType()))\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"label\", df[\"label\"].cast(tp.DoubleType()))"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "# https://spark.apache.org/docs/latest/ml-pipeline.html#example-estimator-transformer-and-param\n",
    "# https://spark.apache.org/docs/latest/ml-features.html#tf-idf\n",
    "\n",
    "tdf = df.select(\n",
    "    concat_ws(\n",
    "        \" \",\n",
    "        df.title,\n",
    "        df.selftext, \n",
    "        df.ocr_text,\n",
    "        df.caption_text\n",
    "    ).alias(\"all_text\"),\n",
    "    \"label\"\n",
    "    )\n"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "train, test = tdf.randomSplit([0.8, 0.2])\n",
    "train.show()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "test.show()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "from pyspark.ml.feature import HashingTF, IDF, Tokenizer\n",
    "\n",
    "tokenizer = Tokenizer(inputCol=\"all_text\", outputCol=\"words\")\n",
    "train_wordsData = tokenizer.transform(train)\n",
    "\n",
    "hashingTF = HashingTF(inputCol=\"words\", outputCol=\"rawFeatures\", numFeatures=2**16)\n",
    "train_featurizedData = hashingTF.transform(train_wordsData)\n",
    "# alternatively, CountVectorizer can also be used to get term frequency vectors\n",
    "\n",
    "idf = IDF(inputCol=\"rawFeatures\", outputCol=\"features\")\n",
    "idfModel = idf.fit(train_featurizedData)\n",
    "train_rescaledData = idfModel.transform(train_featurizedData)\n",
    "# tdf= rescaledData\n",
    "\n",
    "train= train_rescaledData\n",
    "train.show()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "test_wordsData = tokenizer.transform(test)\n",
    "\n",
    "test_featurizedData = hashingTF.transform(test_wordsData)\n",
    "# alternatively, CountVectorizer can also be used to get term frequency vectors\n",
    "test_rescaledData = idfModel.transform(test_featurizedData)\n",
    "# tdf= rescaledData\n",
    "\n",
    "test = test_rescaledData\n",
    "test.show()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "from pyspark.ml.classification import LogisticRegression, OneVsRest\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "\n",
    "lr = LogisticRegression(maxIter=10, tol=1e-6, fitIntercept=True)\n",
    "\n",
    "# instantiate the One Vs Rest Classifier.\n",
    "ovr = OneVsRest(classifier=lr)\n",
    "train = train.withColumn(\"label\", train[\"label\"].cast(tp.DoubleType()))\n",
    "test = test.withColumn(\"label\", test[\"label\"].cast(tp.DoubleType()))\n",
    "# train the multiclass model.\n",
    "ovrModel = ovr.fit(train)\n",
    "\n",
    "# score the model on test data.\n",
    "predictions = ovrModel.transform(test)\n",
    "\n",
    "# obtain evaluator.\n",
    "evaluator = MulticlassClassificationEvaluator(metricName=\"accuracy\")\n",
    "\n",
    "# compute the classification error on test data.\n",
    "accuracy = evaluator.evaluate(predictions)\n",
    "print(\"Test Error = %g\" % (1.0 - accuracy))\n"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "train = train.withColumn(\"label\", train[\"label\"].cast(tp.DoubleType()))\n",
    "test = test.withColumn(\"label\", test[\"label\"].cast(tp.DoubleType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import SQLTransformer\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.ml.classification import LogisticRegression, OneVsRest\n",
    "from pyspark.sql.functions import concat_ws\n",
    "from pyspark.ml.feature import Tokenizer, HashingTF, IDF\n",
    "\n",
    "\n",
    "\"\"\"     df.title,\n",
    "        df.selftext, \n",
    "        df.ocr_text,\n",
    "        df.caption_text \"\"\"\n",
    "\n",
    "\"\"\" sqlTrans = SQLTransformer(\n",
    "    statement=\"SELECT *, concat_ws(\\\" \\\", coalesce(title, ''), \\\n",
    "        coalesce(selftext, ''), \\\n",
    "        coalesce(ocr_text, ''), \\\n",
    "        coalesce(caption_text, '')) AS all_text  FROM __THIS__\"\n",
    ") \"\"\"\n",
    "df = df.withColumn(\"all_text\",concat_ws(\" \", df.title, df.selftext, df.ocr_text, df.caption_text))\n",
    "\n",
    "tokenizer = Tokenizer(inputCol=\"all_text\", outputCol=\"words\")\n",
    "# train_wordsData = tokenizer.transform(train)\n",
    "\n",
    "hashingTF = HashingTF(inputCol=\"words\", outputCol=\"rawFeatures\", numFeatures=2**16)\n",
    "# train_featurizedData = hashingTF.transform(train_wordsData)\n",
    "\n",
    "idf = IDF(inputCol=\"rawFeatures\", outputCol=\"features\")\n",
    "# idfModel = idf.fit(train_featurizedData)\n",
    "# train_rescaledData = idfModel.transform(train_featurizedData)\n",
    "lr = LogisticRegression(maxIter=10, tol=1e-6, fitIntercept=True)\n",
    "\n",
    "ovr = OneVsRest(classifier=lr)\n",
    "\n",
    "\n",
    "pipeline = Pipeline(stages=[tokenizer, hashingTF, idf, ovr])\n",
    "train, test = df.randomSplit([0.8, 0.2]) \n",
    "\n",
    "pipelineModel = pipeline.fit(train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipelineModel.write().overwrite().save(\"./spark/models/ovrModel\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test Error = 0.47\n"
     ]
    }
   ],
   "source": [
    "predictions = pipelineModel.transform(test)\n",
    "evaluator = MulticlassClassificationEvaluator(metricName=\"accuracy\")\n",
    "\n",
    "accuracy = evaluator.evaluate(predictions)\n",
    "print(\"Test Error = %g\" % (1.0 - accuracy))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+-----+----------+\n",
      "|               title|       subreddit|label|prediction|\n",
      "+--------------------+----------------+-----+----------+\n",
      "|    I’m a fat piggy |        GymMemes|  0.0|       3.0|\n",
      "|how did he NEVER ...|        GymMemes|  0.0|       3.0|\n",
      "|      END OF AN ERA.|   footballmemes|  0.0|       1.0|\n",
      "|      END OF AN ERA.|     soccermemes|  0.0|       1.0|\n",
      "|When you need to ...|        GymMemes|  0.0|       3.0|\n",
      "|Which NBA player ...|        Nbamemes|  0.0|       0.0|\n",
      "|Man United fans w...|     soccermemes|  0.0|       0.0|\n",
      "|It really doesn't...|        GymMemes|  0.0|       0.0|\n",
      "|  Sad facts of life |     soccermemes|  0.0|       0.0|\n",
      "|        Wait what 😭|     soccermemes|  0.0|       1.0|\n",
      "|Average depressed...|        GymMemes|  0.0|       1.0|\n",
      "|Cognitive dissona...|        GymMemes|  0.0|       0.0|\n",
      "|Andre Onana saves...|     soccermemes|  0.0|       0.0|\n",
      "|Me after training...|        GymMemes|  0.0|       1.0|\n",
      "|Liverpool fans ri...|     soccermemes|  0.0|       1.0|\n",
      "|A long time ago, ...|        Animemes|  1.0|       3.0|\n",
      "|She’s 500 years o...|        Animemes|  1.0|       1.0|\n",
      "|        Hell yeah 🗿|        GymMemes|  0.0|       3.0|\n",
      "| Spurs in a NLD w...|     soccermemes|  0.0|       0.0|\n",
      "|Imagine thinking ...|TheRightCantMeme|  2.0|       3.0|\n",
      "+--------------------+----------------+-----+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#df=df.drop(\"all_text\")\n",
    "predictions.select(\"title\", \"subreddit\", \"label\", \"prediction\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "100"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictions.count()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "tdf= sqlTrans.transform(df)\n",
    "tdf.show()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "df.printSchema()\n"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {
    "vscode": {
     "languageId": "raw"
    }
   },
   "source": [
    "spark.stop()"
   ]
  }
 ],
 "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.11.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}