Synapse/notebooks/text_clustering/standalone_text_clustering_BERT.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "# Clustering in Spark using BERT Embeddings"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "##  This cell configures the spark session - Do not change"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%configure -f\n",
        "{\n",
        "\"conf\": {\n",
        "     \"spark.rpc.message.maxSize\": 1024,\n",
        "     \"spark.kryoserializer.buffer.max\": \"256m\",\n",
        "     \"spark.driver.maxResultSize\": \"8g\"\n",
        "   }\n",
        "}"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": null,
              "session_id": "11",
              "statement_id": -1,
              "state": "finished",
              "livy_statement_state": "available",
              "queued_time": "2023-03-20T15:02:52.7386286Z",
              "session_start_time": "2023-03-20T15:02:52.7955614Z",
              "execution_start_time": "2023-03-20T15:07:09.5985848Z",
              "execution_finish_time": "2023-03-20T15:07:09.5988088Z",
              "spark_jobs": null,
              "parent_msg_id": "343c1617-8f94-48c6-bebb-fb40062e6647"
            },
            "text/plain": "StatementMeta(, 11, -1, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": 1,
      "metadata": {}
    },
    {
      "cell_type": "markdown",
      "source": [
        "# These are the parameters that need to be changed to your values"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# The blob account url - https://[accountname].blob.core.windows.net\n",
        "account_url = \"https://datadiscoverypipeline2.blob.core.windows.net\"\n",
        "# The blob account name = [accountname]\n",
        "account_name = 'datadiscoverypipeline2'\n",
        "# The blob account key - used to generate a SAS key\n",
        "account_key = ''\n",
        "\n",
        "# The input file name \n",
        "input_filename = 'abfss://share@datadiscoverypipeline2.dfs.core.windows.net/bbcsports/csv/sport_articles.csv'\n",
        "# The number of clusters - this can be automated or start with a guesstimate\n",
        "number_of_clusters = 5\n",
        "# The output directory where the output file will be written to\n",
        "output_directory = 'abfss://share@datadiscoverypipeline2.dfs.core.windows.net/bbcsports/csv/'\n",
        "# The name of the output file\n",
        "output_filename = 'cosine_spacy_max_++_cosine_bert.csv'\n",
        "\n",
        "# The name of the primary ADLS share\n",
        "file_system_name=\"share\"\n",
        "# The directory folders where your files reside  \n",
        "directory_name='bbcsports'  # bbc/videos/\n",
        "\n",
        "# If set to true generate a 3D scatterplot otherwise 2D\n",
        "SCATTER_PLOT_3D = False\n",
        "# If this is set to True then the Coalesce notebook will need to be run to merge the partition files into a single file\n",
        "LOW_MEMORY_MODE = True\n",
        "\n",
        "# Concept Graph - # Get top N most connected nodes   \n",
        "number_of_connected_nodes = 5\n",
        "\n",
        "# Azure SubscriptionId\n",
        "subscription_id=\"\"\n",
        "# AzureML Workspace Resource Group\n",
        "resource_group=\"\"\n",
        "# AzureML Workspace Name\n",
        "workspace_name=\"\""
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "graphx",
              "session_id": "11",
              "statement_id": 1,
              "state": "finished",
              "livy_statement_state": "available",
              "queued_time": "2023-03-20T15:07:52.7353302Z",
              "session_start_time": null,
              "execution_start_time": "2023-03-20T15:08:09.7326708Z",
              "execution_finish_time": "2023-03-20T15:08:09.8984964Z",
              "spark_jobs": null,
              "parent_msg_id": "90ab9bcc-0181-4acd-be30-d55fcee60e4f"
            },
            "text/plain": "StatementMeta(graphx, 11, 1, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": 2,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Track the Experiment in Azure ML"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from azureml.core import Workspace, Experiment, Run\n",
        "import mlflow\n",
        "\n",
        "ws = Workspace(subscription_id=subscription_id, resource_group=resource_group, workspace_name=workspace_name)    \n",
        "mlflow.set_tracking_uri(ws.get_mlflow_tracking_uri())\n",
        "experiment_name = f\"({mssparkutils.runtime.context['notebookname']}_{str(mssparkutils.env.getJobId())})\"\n",
        "mlflow.set_experiment(experiment_name)\n",
        "mlflow.log_param(\"input_filename\", input_filename)\n",
        "mlflow.log_param(\"number_of_clusters\", number_of_clusters)\n",
        "mlflow.log_param(\"output_directory\", output_directory)\n",
        "mlflow.log_param(\"output_filename\", output_filename)\n",
        "mlflow.log_param(\"account_url\", account_url)\n",
        "mlflow.log_param(\"account_name\", account_name)\n",
        "mlflow.log_param(\"file_system_name\", file_system_name)\n",
        "mlflow.log_param(\"directory_name\", directory_name)\n",
        "mlflow.log_param(\"SCATTER_PLOT_3D\", SCATTER_PLOT_3D)\n",
        "mlflow.log_param(\"LOW_MEMORY_MODE\", LOW_MEMORY_MODE)\n",
        "params = {\n",
        "    \"sparkpool\": mssparkutils.runtime.context['sparkpool'],\n",
        "    \"workspace\": mssparkutils.runtime.context['workspace'],\n",
        "    \"notebookname\": mssparkutils.runtime.context['notebookname'],\n",
        "    \"isForPipeline\": mssparkutils.runtime.context['isForPipeline'],\n",
        "    \"pipelinejobid\": mssparkutils.runtime.context['pipelinejobid']\n",
        "}\n",
        "\n",
        "mlflow.log_params(params)\n",
        "mlflow.pyspark.ml.autolog()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.ml.feature import HashingTF, IDF, Tokenizer, CountVectorizer, StopWordsRemover, PCA, RegexTokenizer\n",
        "from pyspark.ml.clustering import LDA, KMeans, BisectingKMeans\n",
        "from pyspark.ml import Pipeline\n",
        "from pyspark import SparkContext, SparkConf\n",
        "import sys\n",
        "from pyspark.sql.functions import udf, col, size\n",
        "from pyspark.sql.types import StringType\n",
        "from pyspark.sql import SparkSession\n",
        "import ntpath\n",
        "import os\n",
        "\n",
        "from pyspark.ml.feature import StopWordsRemover, RFormula\n",
        "from pyspark.sql.functions import udf\n",
        "from pyspark.sql.types import StringType, IntegerType\n",
        "from transformers import BertTokenizer, BertModel\n",
        "from string import punctuation\n",
        "from pyspark.sql.functions import udf,col,lit\n",
        "from pyspark.ml.linalg import Vectors, VectorUDT\n",
        "from pyspark.sql.types import ArrayType,StringType, FloatType\n",
        "\n",
        "from itertools import combinations\n",
        "from operator import itemgetter\n",
        "\n",
        "from graphframes import *\n",
        "from pyspark.sql.functions import monotonically_increasing_id, lit\n",
        "\n",
        "\n",
        "global nlp\n",
        "\n",
        "model_name = \"bert-large-uncased\"\n",
        "\n",
        "import numpy as np\n",
        "from sklearn.feature_extraction.text import TfidfVectorizer\n",
        "\n",
        "# replace this with your folder data\n",
        "df = spark.read.load(input_filename, header=True, format='csv')\n"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "graphx",
              "session_id": "11",
              "statement_id": 2,
              "state": "finished",
              "livy_statement_state": "available",
              "queued_time": "2023-03-20T15:08:15.3214462Z",
              "session_start_time": null,
              "execution_start_time": "2023-03-20T15:08:15.4479632Z",
              "execution_finish_time": "2023-03-20T15:09:03.4283592Z",
              "spark_jobs": null,
              "parent_msg_id": "7533acb7-0b43-4ff9-ab82-b6a7c339ced2"
            },
            "text/plain": "StatementMeta(graphx, 11, 2, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": 3,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Load and initialise the BERT model"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "class Models:\n",
        "    def __init__(self, text_model, tokenizer=None, model=None):\n",
        "        self.tokenizer = tokenizer\n",
        "        self.model = model\n",
        "\n",
        "    def load_text_model(self, text_model):\n",
        "        self.tokenizer = BertTokenizer.from_pretrained(text_model) \n",
        "        self.model = BertModel.from_pretrained(text_model)\n",
        "\n",
        "# Initialise BERT model\n",
        "text_model = Models(text_model=None)\n",
        "text_model.load_text_model(model_name)\n"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "graphx",
              "session_id": "11",
              "statement_id": 3,
              "state": "finished",
              "livy_statement_state": "available",
              "queued_time": "2023-03-20T15:14:48.0811479Z",
              "session_start_time": null,
              "execution_start_time": "2023-03-20T15:14:48.2156255Z",
              "execution_finish_time": "2023-03-20T15:15:07.9712907Z",
              "spark_jobs": null,
              "parent_msg_id": "0b231f71-de61-4b09-abff-192e85ae479c"
            },
            "text/plain": "StatementMeta(graphx, 11, 3, Finished, Available)"
          },
          "metadata": {}
        },
        {
          "output_type": "display_data",
          "data": {
            "text/plain": "Downloading (…)solve/main/vocab.txt:   0%|          | 0.00/232k [00:00<?, ?B/s]",
            "application/vnd.jupyter.widget-view+json": {
              "version_major": 2,
              "version_minor": 0,
              "model_id": "b0f01d6f-19a0-46c2-ad7d-7282f87de1e6"
            }
          },
          "metadata": {}
        },
        {
          "output_type": "display_data",
          "data": {
            "text/plain": "Downloading (…)okenizer_config.json:   0%|          | 0.00/28.0 [00:00<?, ?B/s]",
            "application/vnd.jupyter.widget-view+json": {
              "version_major": 2,
              "version_minor": 0,
              "model_id": "7781f6c0-6c5e-4a6e-9baf-3d37dc3ea8eb"
            }
          },
          "metadata": {}
        },
        {
          "output_type": "display_data",
          "data": {
            "text/plain": "Downloading (…)lve/main/config.json:   0%|          | 0.00/571 [00:00<?, ?B/s]",
            "application/vnd.jupyter.widget-view+json": {
              "version_major": 2,
              "version_minor": 0,
              "model_id": "6989af71-462e-48c7-b3b9-0bd951aea6d3"
            }
          },
          "metadata": {}
        },
        {
          "output_type": "display_data",
          "data": {
            "text/plain": "Downloading pytorch_model.bin:   0%|          | 0.00/1.34G [00:00<?, ?B/s]",
            "application/vnd.jupyter.widget-view+json": {
              "version_major": 2,
              "version_minor": 0,
              "model_id": "c19b9f1d-a807-4b79-af3f-45551593c5ec"
            }
          },
          "metadata": {}
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "Some weights of the model checkpoint at bert-large-uncased were not used when initializing BertModel: ['cls.predictions.transform.dense.bias', 'cls.seq_relationship.bias', 'cls.predictions.transform.LayerNorm.weight', 'cls.predictions.transform.dense.weight', 'cls.predictions.decoder.weight', 'cls.predictions.transform.LayerNorm.bias', 'cls.seq_relationship.weight', 'cls.predictions.bias']\n- This IS expected if you are initializing BertModel from the checkpoint of a model trained on another task or with another architecture (e.g. initializing a BertForSequenceClassification model from a BertForPreTraining model).\n- This IS NOT expected if you are initializing BertModel from the checkpoint of a model that you expect to be exactly identical (initializing a BertForSequenceClassification model from a BertForSequenceClassification model).\n"
          ]
        },
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "graphx",
              "session_id": "11",
              "statement_id": 4,
              "state": "finished",
              "livy_statement_state": "available",
              "queued_time": "2023-03-20T15:14:49.8871534Z",
              "session_start_time": null,
              "execution_start_time": "2023-03-20T15:15:08.1338096Z",
              "execution_finish_time": "2023-03-20T15:15:08.3034053Z",
              "spark_jobs": null,
              "parent_msg_id": "79667d20-d301-4eb8-8150-af10757a1369"
            },
            "text/plain": "StatementMeta(graphx, 11, 4, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": 4,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Using BERT for Tokenisation, Vectorisation via custom functions and udf"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def get_features(text):     \n",
        "    inputs = text_model.tokenizer(text[0:1000], return_tensors=\"pt\")\n",
        "    outputs = text_model.model(**inputs)\n",
        "    last_hidden_states = outputs.last_hidden_state\n",
        "    last_hidden_states = last_hidden_states.detach().numpy()\n",
        "  \n",
        "    return Vectors.dense(last_hidden_states[0][0])\n",
        "\n",
        "# Get features using the BERT model\n",
        "udf_text = udf(get_features, VectorUDT())\n",
        "df = df.withColumn(\"bert_features\", udf_text(col(\"text\"))) "
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "graphx",
              "session_id": "11",
              "statement_id": 5,
              "state": "finished",
              "livy_statement_state": "available",
              "queued_time": "2023-03-20T15:15:12.5410996Z",
              "session_start_time": null,
              "execution_start_time": "2023-03-20T15:15:12.6711833Z",
              "execution_finish_time": "2023-03-20T15:15:17.9928698Z",
              "spark_jobs": null,
              "parent_msg_id": "efba8b62-6c41-4a15-b457-81913d5aaf28"
            },
            "text/plain": "StatementMeta(graphx, 11, 5, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": 5,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Apply PCA and Kmeans clustering"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Apply PCA and Kmeans in a pipeline\n",
        "pca = PCA(k=20, inputCol=\"bert_features\")\n",
        "pca.setOutputCol(\"features\")\n",
        "\n",
        "if SCATTER_PLOT_3D:\n",
        "  pca_2 = PCA(k=3, inputCol=\"features\")\n",
        "else:  \n",
        "  pca_2 = PCA(k=2, inputCol=\"features\")\n",
        "  \n",
        "pca_2.setOutputCol(\"pca_scatterplot_features\")\n",
        "\n",
        "kmeans = KMeans(k=number_of_clusters, seed=42, initMode=\"k-means||\", distanceMeasure=\"euclidean\")\n",
        "\n",
        "pipeline = Pipeline(stages=[ pca, kmeans, pca_2])\n",
        "model = pipeline.fit(df)\n",
        "df_coords = model.transform(df)"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "graphx",
              "session_id": "11",
              "statement_id": 6,
              "state": "finished",
              "livy_statement_state": "available",
              "queued_time": "2023-03-20T15:15:21.4439515Z",
              "session_start_time": null,
              "execution_start_time": "2023-03-20T15:15:21.5982979Z",
              "execution_finish_time": "2023-03-20T15:44:38.6389396Z",
              "spark_jobs": null,
              "parent_msg_id": "4152e9cc-f3d7-4eb8-87aa-37802fb510b9"
            },
            "text/plain": "StatementMeta(graphx, 11, 6, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": 6,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Retrieve Data from Blob Storage"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from azure.identity import DefaultAzureCredential\n",
        "from datetime import datetime, timedelta\n",
        "from azure.storage.blob import BlobServiceClient, generate_container_sas, BlobSasPermissions\n",
        "token_credential = DefaultAzureCredential()\n",
        "\n",
        "blob_service_client = BlobServiceClient(\n",
        "    account_url=account_url,\n",
        "    credential=token_credential\n",
        ")\n",
        "\n",
        "from azure.storage.filedatalake import DataLakeServiceClient, generate_directory_sas\n",
        "SAS = generate_directory_sas(\n",
        "        account_name=account_name,\n",
        "        file_system_name=file_system_name,\n",
        "        directory_name=directory_name,\n",
        "        credential=account_key,\n",
        "        permission=BlobSasPermissions(read=True),\n",
        "        expiry=datetime.utcnow() + timedelta(days=100))\n",
        "\n",
        "SAS_key  = \"?\" + SAS\n",
        "\n",
        "storage_path = os.path.join(account_url, file_system_name, directory_name)\n",
        "SAS_path = []\n",
        "\n",
        "def build_sas_path(row):\n",
        "    file_name = ntpath.basename(row)\n",
        "    return account_url + \"/\" + file_system_name + \"/\" + directory_name + \"/\" + file_name + SAS_key\n",
        "\n",
        "udf_build_sas_path = udf(build_sas_path, StringType())\n",
        "\n",
        "def get_X(row):\n",
        "    return str(row.values[0])\n",
        "\n",
        "def get_Y(row):\n",
        "    return str(row.values[1])\n",
        "\n",
        "def get_Z(row):\n",
        "    return str(row.values[2])\n",
        "\n",
        "def join_text(row):\n",
        "    return \"\".join(row)\n",
        "\n",
        "def remove_separator(row):\n",
        "    return row.replace(\",\", \"\").replace(\"\\\\\", \"\")\n",
        "\n",
        "udf_get_X = udf(get_X, StringType())\n",
        "udf_get_Y = udf(get_Y, StringType())\n",
        "udf_get_Z = udf(get_Z, StringType())\n",
        "udf_join_text = udf(join_text, StringType())\n",
        "\n",
        "\n",
        "    \n",
        "udf_remove_separator = udf(remove_separator, StringType())\n",
        "\n",
        "\n",
        "if SCATTER_PLOT_3D:\n",
        "    df_coords = df_coords.withColumn(\"blob_path\", udf_build_sas_path(df_coords.filename)).withColumn(\"X\", udf_get_X(df_coords.pca_scatterplot_features).cast('string')).withColumn(\"Y\", udf_get_Y(df_coords.pca_scatterplot_features).cast('string')).withColumn(\"Z\", udf_get_Z(df_coords.pca_scatterplot_features).cast('string'))\n",
        "else:\n",
        "    df_coords = df_coords.withColumn(\"blob_path\", udf_build_sas_path(df_coords.filename)).withColumn(\"X\", udf_get_X(df_coords.pca_scatterplot_features).cast('string')).withColumn(\"Y\", udf_get_Y(df_coords.pca_scatterplot_features).cast('string'))\n",
        "\n",
        "df_graph = df_coords\n",
        "df_coords = df_coords.drop('pca_features', 'pca_scatterplot_features', 'features', 'bert_features')"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "spark_pool": "graphx",
              "session_id": "11",
              "statement_id": 11,
              "state": "finished",
              "livy_statement_state": "available",
              "queued_time": "2023-03-20T15:50:28.0666493Z",
              "session_start_time": null,
              "execution_start_time": "2023-03-20T15:50:28.1858561Z",
              "execution_finish_time": "2023-03-20T15:50:28.3509735Z",
              "spark_jobs": null,
              "parent_msg_id": "a1fe13fd-96b5-433b-9986-650b10397b57"
            },
            "text/plain": "StatementMeta(graphx, 11, 11, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": 11,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Save the outcome with two operations (usual or coalesce)"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "if LOW_MEMORY_MODE:\n",
        "    df_coords.write.mode('overwrite').options(header='true').csv(os.path.join(output_directory, output_filename))\n",
        "else:\n",
        "    df1 = df_coords.filter((df_coords.blob_path != 'blob_path'))\n",
        "    df1.coalesce(1).write.mode('overwrite').options(header='true').csv(os.path.join(output_directory, output_filename))\n",
        "\n",
        "mlflow.pyspark.ml.mlflow.end_run()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Optional: Add Azure Cognitive Search"
      ],
      "metadata": {}
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Add Search Parameters"
      ],
      "metadata": {}
    },
    {
      "cell_type": "code",
      "source": [
        "# Azure Search Admin Key\n",
        "search_admin_key = \"\"\n",
        "# The name of the search service\n",
        "search_service_name = \"\"\n",
        "# The Azure Search Query Key\n",
        "search_query_key = \"\""
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {}
    },
    {
      "cell_type": "code",
      "source": [
        "from synapse.ml.cognitive import *\n",
        "from pyspark.sql.functions import monotonically_increasing_id, lit\n",
        "\n",
        "df = df.drop(\"_c0\")\n",
        "\n",
        "(\n",
        "    df.withColumn(\"key\", monotonically_increasing_id().cast(\"string\"))\n",
        "    .withColumn(\"SearchAction\", lit(\"upload\"))\n",
        "    .writeToAzureSearch(\n",
        "        subscriptionKey=search_admin_key,\n",
        "        actionCol=\"SearchAction\",\n",
        "        serviceName=search_service_name,\n",
        "        indexName=experiment_name,  # Defaults to the notebook name\n",
        "        keyCol=\"key\",\n",
        "    )\n",
        ")"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {}
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Search the generated Azure Search Index"
      ],
      "metadata": {}
    },
    {
      "cell_type": "code",
      "source": [
        "import requests\n",
        "from IPython.display import display, Markdown\n",
        "\n",
        "term_to_search_for = \"covid\"\n",
        "\n",
        "url = \"https://{}.search.windows.net/indexes/{}/docs/search?api-version=2019-05-06\".format(\n",
        "    search_service_name, experiment_name\n",
        "\n",
        ")\n",
        "jdata = requests.post(url, json={\"search\": term_to_search_for}, headers={\"api-key\": search_query_key}).json()\n",
        "\n",
        "for doc in jdata['value']:\n",
        "    display(Markdown(f'**Search Score {doc[\"@search.score\"]}** Document {doc[\"filename\"]}'))\n",
        "    display(Markdown(f'{doc[\"text\"]}'))\n"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {}
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Implement Semantic Search"
      ],
      "metadata": {}
    },
    {
      "cell_type": "markdown",
      "source": [
        "1) [Enable Semantic Search](https://docs.microsoft.com/en-us/azure/search/semantic-search-overview#enable-semantic-search) on your search instance\n",
        "\n",
        "2) [Configure Semantic Search](https://docs.microsoft.com/en-us/azure/search/semantic-how-to-query-request?tabs=semanticConfiguration%2Cportal#create-a-semantic-configuration)"
      ],
      "metadata": {}
    },
    {
      "cell_type": "code",
      "source": [
        "term_to_search_for = \"Whose thumb was fractured?\"\n",
        "\n",
        "url = \"https://{}.search.windows.net/indexes/{}/docs/search?api-version=2021-04-30-Preview\".format(\n",
        "    search_service_name, experiment_name\n",
        ")\n",
        "jdata = requests.post(url, json={\"search\": term_to_search_for, \"queryType\": \"semantic\", \"semanticConfiguration\": \"config\", \"queryLanguage\": \"en-us\", \"answers\": \"extractive|count-3\",\n",
        "\"captions\": \"extractive|highlight-true\",  \"highlightPreTag\": \"<mark>\",\"highlightPostTag\": \"</mark>\"}, headers={\"api-key\": search_query_key}).json()\n",
        "\n",
        "for doc in jdata['value']:\n",
        "    display(Markdown(f'**Search Score {doc[\"@search.score\"]}** **Search rerankerScore Score {doc[\"@search.rerankerScore\"]}** Document {doc[\"filename\"]}'))\n",
        "    display(Markdown(f'@search.captions {doc[\"@search.captions\"]}'))\n"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {}
    },
    {
      "attachments": {},
      "cell_type": "markdown",
      "source": [
        "# Optional:  Build the Concept Graph using GraphFrames"
      ],
      "metadata": {}
    },
    {
      "cell_type": "code",
      "source": [
        "lst_text = df_graph.select('stopWordsRemovedTokens').rdd.flatMap(lambda x: x).collect()\n",
        "\n",
        "G = nx.Graph()\n",
        "\n",
        "lst_source_node = []\n",
        "lst_source_node_weight = []\n",
        "lst_source_node_label = []\n",
        "lst_target_node = []\n",
        "lst_target_node_weight = []\n",
        "lst_target_node_label = []\n",
        "lst_source_url = []\n",
        "lst_target_url = []\n",
        "lst_edge_weight = []\n",
        "lst_edge_colour_weight = []\n",
        "\n",
        "dict_nodes = {}\n",
        "\n",
        "from itertools import combinations\n",
        "\n",
        "for i, row in enumerate(lst_text):\n",
        "    \n",
        "    combos = list(combinations(row, 2))\n",
        "   \n",
        "    for c in combos:\n",
        "        # First update edge weights\n",
        "        if (c[0] + \"_\" + c[1] not in dict_nodes) and (c[1] + \"_\" + c[0] not in dict_nodes):\n",
        "            dict_nodes[c[0] + \"_\" + c[1]] = 1 # initialise anc create first combo\n",
        "        elif c[0] + \"_\" + c[1] in dict_nodes:\n",
        "            dict_nodes[c[0] + \"_\" + c[1]] += 1\n",
        "        elif c[1] + \"_\" + c[0] in dict_nodes:\n",
        "            dict_nodes[c[1] + \"_\" + c[0]] += 1\n",
        "\n",
        "    for c in combos:\n",
        "        lst_source_node.append(c[0])\n",
        "        G.add_node(c[0])#, label=ntpath.basename(df['path'].iloc[i]), x=row.X, y=row.Y)\n",
        "        lst_target_node.append(c[1])\n",
        "        G.add_node(c[1])#), label=ntpath.basename(df['path'].iloc[i]), x=row.X, y=row.Y)\n",
        "        if c[0] + \"_\" + c[1] in dict_nodes:\n",
        "            lst_edge_weight.append(dict_nodes[c[0] + \"_\" + c[1]])\n",
        "            lst_source_node_weight.append(dict_nodes[c[0] + \"_\" + c[1]])\n",
        "            lst_target_node_weight.append(dict_nodes[c[0] + \"_\" + c[1]])\n",
        "            G.add_edge(c[0], c[1])\n",
        "        else:\n",
        "            lst_edge_weight.append(dict_nodes[c[1] + \"_\" + c[0]])\n",
        "            lst_source_node_weight.append(dict_nodes[c[1] + \"_\" + c[0]])\n",
        "            lst_target_node_weight.append(dict_nodes[c[1] + \"_\" + c[0]])\n",
        "            #lst_source_node_label.append()\n",
        "\n",
        "topn = dict(sorted(dict_nodes.items(), key = itemgetter(1), reverse = True)[:number_of_connected_nodes])\n",
        "\n",
        "# Assign edge weight colour\n",
        "for key in zip(lst_source_node, lst_target_node):\n",
        "    \n",
        "    if key[0] + \"_\" + key[1] in topn or key[1] + \"_\" + key[0] in topn:\n",
        "        lst_edge_colour_weight.append(\"red\")\n",
        "    else:\n",
        "        lst_edge_colour_weight.append(\"black\")\n",
        "\n",
        "\n",
        "# Create the Graph RDD\n",
        "columns = ['source', 'target', 'source_node_weight', 'target_node_weight', 'edge_weight', 'edge_colour']\n",
        "df_concept_graph = spark.createDataFrame(zip(lst_source_node, lst_target_node, lst_source_node_weight, lst_target_node_weight, lst_edge_weight, lst_edge_colour_weight), columns)\n",
        "  \n"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {}
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Amend this section to build your concept graph"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "lst_text = df_graph.select('text').rdd.flatMap(lambda x: x).collect()\n",
        "\n",
        "lst_source_node = []\n",
        "lst_source_node_weight = []\n",
        "lst_source_node_label = []\n",
        "lst_target_node = []\n",
        "lst_target_node_weight = []\n",
        "lst_target_node_label = []\n",
        "lst_source_url = []\n",
        "lst_target_url = []\n",
        "lst_edge_weight = []\n",
        "lst_edge_colour_weight = []\n",
        "\n",
        "lst_g_nodes = []\n",
        "lst_g_edges = []\n",
        "\n",
        "dict_nodes = {}\n",
        "\n",
        "from itertools import combinations\n",
        "\n",
        "for i, row in enumerate(lst_text):\n",
        "    \n",
        "    combos = list(combinations(row, 2))\n",
        "   \n",
        "    for c in combos:\n",
        "        # First update edge weights\n",
        "        if (c[0] + \"_\" + c[1] not in dict_nodes) and (c[1] + \"_\" + c[0] not in dict_nodes):\n",
        "            dict_nodes[c[0] + \"_\" + c[1]] = 1 # initialise and create first combo\n",
        "        elif c[0] + \"_\" + c[1] in dict_nodes:\n",
        "            dict_nodes[c[0] + \"_\" + c[1]] += 1\n",
        "        elif c[1] + \"_\" + c[0] in dict_nodes:\n",
        "            dict_nodes[c[1] + \"_\" + c[0]] += 1\n",
        "\n",
        "    for c in combos:\n",
        "        lst_source_node.append(c[0])\n",
        "        lst_g_nodes.append((c[0], c[0]))\n",
        "        lst_target_node.append((c[1]))\n",
        "        lst_g_nodes.append((c[1], c[1]))\n",
        "        if c[0] + \"_\" + c[1] in dict_nodes:\n",
        "            lst_edge_weight.append(dict_nodes[c[0] + \"_\" + c[1]])\n",
        "            lst_source_node_weight.append(dict_nodes[c[0] + \"_\" + c[1]])\n",
        "            lst_target_node_weight.append(dict_nodes[c[0] + \"_\" + c[1]])\n",
        "            lst_g_edges.append((c[0],c[1], \"related\"))\n",
        "        else:\n",
        "            lst_edge_weight.append(dict_nodes[c[1] + \"_\" + c[0]])\n",
        "            lst_source_node_weight.append(dict_nodes[c[1] + \"_\" + c[0]])\n",
        "            lst_target_node_weight.append(dict_nodes[c[1] + \"_\" + c[0]])\n",
        "\n",
        "topn = dict(sorted(dict_nodes.items(), key = itemgetter(1), reverse = True)[:number_of_connected_nodes])\n",
        "\n",
        "# Assign edge weight colour\n",
        "for key in zip(lst_source_node, lst_target_node):\n",
        "    \n",
        "    if key[0] + \"_\" + key[1] in topn or key[1] + \"_\" + key[0] in topn:\n",
        "        lst_edge_colour_weight.append(\"red\")\n",
        "    else:\n",
        "        lst_edge_colour_weight.append(\"black\")\n",
        "\n",
        "\n",
        "# Create the Graph RDD\n",
        "columns = ['source', 'target', 'source_node_weight', 'target_node_weight', 'edge_weight', 'edge_colour']\n",
        "df_concept_graph = spark.createDataFrame(zip(lst_source_node, lst_target_node, lst_source_node_weight, lst_target_node_weight, lst_edge_weight, lst_edge_colour_weight), columns)\n",
        "  \n",
        "# Create a Vertex DataFrame with unique ID column \"id\"\n",
        "v = sqlContext.createDataFrame(lst_g_nodes, [\"id\", \"name\"])\n",
        "# Create an Edge DataFrame with \"src\" and \"dst\" columns\n",
        "e = sqlContext.createDataFrame(lst_g_edges, [\"src\", \"dst\", \"relationship\"])"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Show degree connectivity"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from graphframes import GraphFrame\n",
        "g = GraphFrame(v, e)\n",
        "\n",
        "# Query: Get in-degree of each vertex.\n",
        "df_degree = g.inDegrees\n",
        "df_degree.sort(['inDegree'], ascending=False).show()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Run PageRank "
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Query: Count the number of \"follow\" connections in the graph.\n",
        "g.edges.filter(\"relationship = 'relationship'\").count()\n",
        "\n",
        "# Run PageRank algorithm, and show results.\n",
        "results = g.pageRank(resetProbability=0.01, maxIter=20)\n",
        "results.vertices.select(\"id\", \"pagerank\").show()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "attachments": {},
      "cell_type": "markdown",
      "source": [
        "# Save the outcome with two operations (usual or coalesce)"
      ],
      "metadata": {}
    },
    {
      "cell_type": "code",
      "source": [
        "df_concept_graph.write.mode('overwrite').options(header='true').csv(os.path.join(output_directory, output_filename[:-4] + \"concept_graph.csv\"))"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {}
    }
  ],
  "metadata": {
    "kernelspec": {
      "name": "synapse_pyspark",
      "language": "Python",
      "display_name": "Synapse PySpark"
    },
    "language_info": {
      "name": "python"
    },
    "kernel_info": {
      "name": "synapse_pyspark"
    },
    "description": null,
    "save_output": true,
    "synapse_widget": {
      "version": "0.1",
      "state": {}
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}