Synapse/notebooks/text_summarisation/standalone_text_summarisation.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "Copyright (c) Microsoft Corporation. All rights reserved.\n",
        "\n",
        "Licensed under the MIT License.\n",
        "\n",
        "## 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",
        "   }\n",
        "}"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "execution_finish_time": "2022-06-23T14:49:36.3159953Z",
              "execution_start_time": "2022-06-23T14:49:36.3156161Z",
              "livy_statement_state": "available",
              "queued_time": "2022-06-23T14:45:30.2410402Z",
              "session_id": 45,
              "session_start_time": "2022-06-23T14:45:30.6051645Z",
              "spark_pool": null,
              "state": "finished",
              "statement_id": -1
            },
            "text/plain": "StatementMeta(, 45, -1, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "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 input file directory\n",
        "input_directory = \"abfss://share@datadiscoverypipeline.dfs.core.windows.net/bbcsports\"\n",
        "# The output directory where the output file will be written to\n",
        "output_directory = 'abfss://share@datadiscoverypipeline.dfs.core.windows.net/videos_outputs/'\n",
        "# The name of the output file\n",
        "output_filename = 'bbc_text_summarisation.csv'\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",
        "# AML Experiment tracking\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": {
              "execution_finish_time": "2022-06-23T14:49:56.9731899Z",
              "execution_start_time": "2022-06-23T14:49:56.8223487Z",
              "livy_statement_state": "available",
              "queued_time": "2022-06-23T14:45:30.3015048Z",
              "session_id": 45,
              "session_start_time": null,
              "spark_pool": "DataDiscovery",
              "state": "finished",
              "statement_id": 1
            },
            "text/plain": "StatementMeta(DataDiscovery, 45, 1, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Track the Experiment in Azure ML"
      ],
      "metadata": {}
    },
    {
      "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(\"output_directory\", output_directory)\n",
        "mlflow.log_param(\"output_filename\", output_filename)\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": {}
    },
    {
      "cell_type": "code",
      "source": [
        "from  pyspark.sql.functions import input_file_name\n",
        "from pyspark.sql.types import StructType, StructField, StringType, ArrayType, FloatType, IntegerType\n",
        "from pyspark.sql.functions import spark_partition_id\n",
        "from pyspark.ml.linalg import Vectors, VectorUDT\n",
        "from transformers import PegasusTokenizer, PegasusForConditionalGeneration\n",
        "import os\n",
        "from graphframes import *\n",
        "\n",
        "df = spark.read.text(input_directory, wholetext=True)   \n",
        "df = df.withColumn(\"filename\", input_file_name())"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "execution_finish_time": "2022-06-23T14:53:03.1383495Z",
              "execution_start_time": "2022-06-23T14:53:02.074471Z",
              "livy_statement_state": "available",
              "queued_time": "2022-06-23T14:53:01.9750749Z",
              "session_id": 45,
              "session_start_time": null,
              "spark_pool": "DataDiscovery",
              "state": "finished",
              "statement_id": 7
            },
            "text/plain": "StatementMeta(DataDiscovery, 45, 7, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Run the abstractive summarisation process"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import numpy as np\n",
        "import torch\n",
        "\n",
        "\n",
        "def pegasus_summarise_text(df):\n",
        "\n",
        "    model = PegasusForConditionalGeneration.from_pretrained(\"google/pegasus-xsum\")\n",
        "    tokenizer = PegasusTokenizer.from_pretrained(\"google/pegasus-xsum\")\n",
        "\n",
        "    text_abs = []\n",
        "    text_abs = df.value.values.tolist()\n",
        "\n",
        "    summaries = []\n",
        "    for c in text_abs:\n",
        "        inputs = tokenizer(c, max_length=512, return_tensors=\"pt\", truncation=True)\n",
        "        summary_ids = model.generate(inputs[\"input_ids\"])\n",
        "        summary = tokenizer.batch_decode(summary_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False)[0]\n",
        "        summaries.append(summary)\n",
        "\n",
        "    return_df = (\n",
        "        df[[\"filename\"]]\n",
        "        .assign(value=summaries)\n",
        "    )\n",
        "    return return_df"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "execution_finish_time": "2022-06-23T14:53:05.5145479Z",
              "execution_start_time": "2022-06-23T14:53:05.3677624Z",
              "livy_statement_state": "available",
              "queued_time": "2022-06-23T14:53:05.1836356Z",
              "session_id": 45,
              "session_start_time": null,
              "spark_pool": "DataDiscovery",
              "state": "finished",
              "statement_id": 8
            },
            "text/plain": "StatementMeta(DataDiscovery, 45, 8, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Define the schema"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "summary_schema = StructType(\n",
        "    [\n",
        "        StructField(\"filename\", StringType(), True),\n",
        "        StructField(\"value\", StringType(), True)\n",
        "    ]\n",
        ")\n",
        "\n",
        "summary_df = (\n",
        "    df\n",
        "    .groupBy(\"filename\")\n",
        "    .applyInPandas(pegasus_summarise_text, summary_schema)\n",
        ")\n"
      ],
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "execution_finish_time": "2022-06-23T14:53:10.5300885Z",
              "execution_start_time": "2022-06-23T14:53:10.3610363Z",
              "livy_statement_state": "available",
              "queued_time": "2022-06-23T14:53:09.9249217Z",
              "session_id": 45,
              "session_start_time": null,
              "spark_pool": "DataDiscovery",
              "state": "finished",
              "statement_id": 9
            },
            "text/plain": "StatementMeta(DataDiscovery, 45, 9, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "execution_count": null,
      "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": [
        "df_graph = summary_df\n",
        "\n",
        "if LOW_MEMORY_MODE:\n",
        "    summary_df.write.mode('overwrite').options(header='true').csv(os.path.join(output_directory, output_filename))\n",
        "else:\n",
        "    summary_df.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": [
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.livy.statement-meta+json": {
              "execution_finish_time": "2022-06-23T15:02:47.1350912Z",
              "execution_start_time": "2022-06-23T14:53:13.2108622Z",
              "livy_statement_state": "available",
              "queued_time": "2022-06-23T14:53:13.1127242Z",
              "session_id": 45,
              "session_start_time": null,
              "spark_pool": "DataDiscovery",
              "state": "finished",
              "statement_id": 10
            },
            "text/plain": "StatementMeta(DataDiscovery, 45, 10, Finished, Available)"
          },
          "metadata": {}
        }
      ],
      "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": {}
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Optional:  Build the Concept Graph using GraphFrames"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Optional: Amend this section to build your concept graph"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "lst_text = df_graph.select('value').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
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Save the outcome with two operations (usual or coalesce)"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "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": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    }
  ],
  "metadata": {
    "kernelspec": {
      "name": "synapse_pyspark",
      "display_name": "python"
    },
    "language_info": {
      "name": "python"
    },
    "description": null,
    "save_output": true,
    "synapse_widget": {
      "version": "0.1",
      "state": {}
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}