INFODISv1.1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ff13171f-e9cc-463b-887e-81ee918e6b5b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, collect_list, trim, array, udf\n",
    "from pyspark.sql.types import IntegerType, ArrayType, StringType\n",
    "from pyspark.ml.feature import MinHashLSH, HashingTF\n",
    "from collections import defaultdict\n",
    "import logging\n",
    "import tempfile\n",
    "import os\n",
    "import editdistance\n",
    "from tqdm import tqdm\n",
    "\n",
    "# Setup logging\n",
    "logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')\n",
    "\n",
    "warehouse_location = tempfile.mkdtemp()\n",
    "\n",
    "def set_env_vars():\n",
    "    python_path = os.path.join(os.environ['CONDA_PREFIX'], 'python.exe')\n",
    "    os.environ['PYSPARK_PYTHON'] = python_path\n",
    "    os.environ['PYSPARK_DRIVER_PYTHON'] = python_path\n",
    "\n",
    "# Set environment variables\n",
    "set_env_vars()\n",
    "\n",
    "# Initialize Spark session\n",
    "spark = SparkSession.builder \\\n",
    "    .appName(\"ProcessGrouping\") \\\n",
    "    .master(\"local[*]\") \\\n",
    "    .config(\"spark.sql.warehouse.dir\", f\"file:///{warehouse_location}\") \\\n",
    "    .config(\"spark.executor.memory\", \"16g\") \\\n",
    "    .config(\"spark.driver.memory\", \"16g\") \\\n",
    "    .config(\"spark.sql.execution.arrow.pyspark.enabled\", \"true\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "# Define paths\n",
    "log_file_path = \"C:/Users/dcave/Documents/DIS_project/DIS_project/data/server_log.txt\"\n",
    "part1_output_path = 'C:/Users/dcave/Documents/DIS_project/part1Output.txt'\n",
    "part1_observations_path = 'C:/Users/dcave/Documents/DIS_project/part1Observations.txt'\n",
    "part2_observations_path = 'C:/Users/dcave/Documents/DIS_project/part2Observations.txt'\n",
    "\n",
    "logging.info(\"Starting to read the log file\")\n",
    "\n",
    "# Read log to df\n",
    "log_df = spark.read.option(\"header\", \"false\").csv(log_file_path)\n",
    "log_df = log_df.limit(200000)\n",
    "log_df = log_df.withColumnRenamed(\"_c0\", \"from_server\") \\\n",
    "               .withColumnRenamed(\"_c1\", \"to_server\") \\\n",
    "               .withColumnRenamed(\"_c2\", \"time\") \\\n",
    "               .withColumnRenamed(\"_c3\", \"action\") \\\n",
    "               .withColumnRenamed(\"_c4\", \"process_id\")\n",
    "\n",
    "logging.info(\"Completed reading and renaming columns of the log file\")\n",
    "\n",
    "# Convert columns to appropriate data types\n",
    "log_df = log_df.withColumn(\"time\", col(\"time\").cast(IntegerType())) \\\n",
    "               .withColumn(\"process_id\", col(\"process_id\").cast(IntegerType())) \\\n",
    "               .withColumn(\"action\", trim(col(\"action\")))\n",
    "logging.info(\"Converted columns to appropriate data types\")\n",
    "\n",
    "# Cache the DataFrame after sampling and transformations\n",
    "log_df.cache()\n",
    "logging.info(\"Cached the log DataFrame\")\n",
    "\n",
    "# Group by process_id and collect log entries\n",
    "processes_df = log_df.groupBy(\"process_id\").agg(collect_list(array(col(\"from_server\"), col(\"to_server\"), col(\"time\"), col(\"action\"))).alias(\"events\"))\n",
    "logging.info(\"Grouped by process_id and collected log entries\")\n",
    "\n",
    "# Cache the grouped DataFrame\n",
    "processes_df.cache()\n",
    "logging.info(\"Cached the processes DataFrame\")\n",
    "\n",
    "# Convert events to server sequences ignoring timestamps\n",
    "def concat_events(events):\n",
    "    return [\"{}_{}\".format(event[0], event[1]) for event in events]\n",
    "\n",
    "concat_events_udf = udf(concat_events, ArrayType(StringType()))\n",
    "processes_df = processes_df.withColumn(\"server_sequence\", concat_events_udf(col(\"events\")))\n",
    "\n",
    "# Transform data using HashingTF for MinHashLSH\n",
    "hashingTF = HashingTF(inputCol=\"server_sequence\", outputCol=\"features\", numFeatures=1000)\n",
    "featurized_df = hashingTF.transform(processes_df)\n",
    "logging.info(\"Transformed features using HashingTF\")\n",
    "\n",
    "# Apply MinHash LSH\n",
    "mh = MinHashLSH(inputCol=\"features\", outputCol=\"hashes\", numHashTables=3)\n",
    "model = mh.fit(featurized_df)\n",
    "logging.info(\"Fitted MinHashLSH model\")\n",
    "\n",
    "# Transform data\n",
    "transformed_df = model.transform(featurized_df)\n",
    "logging.info(\"Transformed data using MinHashLSH model\")\n",
    "\n",
    "# Approximate similarity join\n",
    "candidate_pairs = model.approxSimilarityJoin(transformed_df, transformed_df, 0.3, distCol=\"JaccardDistance\") \\\n",
    "    .select(col(\"datasetA.process_id\").alias(\"pid1\"),\n",
    "            col(\"datasetB.process_id\").alias(\"pid2\"),\n",
    "            col(\"JaccardDistance\"))\n",
    "\n",
    "logging.info(\"Calculated Jaccard similarity and generated candidate pairs\")\n",
    "\n",
    "# Filter self-joins\n",
    "candidate_pairs_filtered = candidate_pairs.filter(\"pid1 < pid2\")\n",
    "\n",
    "# Log number of candidate pairs\n",
    "num_candidate_pairs = candidate_pairs_filtered.count()\n",
    "logging.info(f\"Number of candidate pairs generated: {num_candidate_pairs}\")\n",
    "\n",
    "# Collect process events as a dictionary\n",
    "process_events_dict = processes_df.select(\"process_id\", \"server_sequence\", \"events\").rdd.map(lambda row: (row[\"process_id\"], (row[\"server_sequence\"], row[\"events\"]))).collectAsMap()\n",
    "process_events_broadcast = spark.sparkContext.broadcast(process_events_dict)\n",
    "\n",
    "# Function to calculate edit distance using the editdistance package\n",
    "def calculate_edit_distance(pid1, pid2, process_events_dict):\n",
    "    seq1 = process_events_dict[pid1][0]\n",
    "    seq2 = process_events_dict[pid2][0]\n",
    "    return editdistance.eval(seq1, seq2)\n",
    "\n",
    "# Filter candidate pairs for part 2 (edit distance < 4)\n",
    "def filter_candidates_part2(row):\n",
    "    pid1, pid2 = row[\"pid1\"], row[\"pid2\"]\n",
    "    edit_dist = calculate_edit_distance(pid1, pid2, process_events_broadcast.value)\n",
    "    if edit_dist < 4:\n",
    "        return (pid1, pid2)\n",
    "    return None\n",
    "\n",
    "filtered_candidates_part2_rdd = candidate_pairs_filtered.rdd.map(filter_candidates_part2).filter(lambda x: x is not None)\n",
    "filtered_candidates_part2 = filtered_candidates_part2_rdd.collect()\n",
    "logging.info(\"Filtered candidates for part 2 based on edit distance < 4\")\n",
    "\n",
    "# Further filter the part 2 candidates for part 1 (edit distance < 2)\n",
    "def filter_candidates_part1(row):\n",
    "    pid1, pid2 = row\n",
    "    edit_dist = calculate_edit_distance(pid1, pid2, process_events_broadcast.value)\n",
    "    if edit_dist < 2:\n",
    "        return (pid1, pid2)\n",
    "    return None\n",
    "\n",
    "filtered_candidates_part1_rdd = spark.sparkContext.parallelize(filtered_candidates_part2).map(filter_candidates_part1).filter(lambda x: x is not None)\n",
    "filtered_candidates_part1 = filtered_candidates_part1_rdd.collect()\n",
    "logging.info(\"Filtered candidates for part 1 based on edit distance < 2\")\n",
    "\n",
    "# Group similar pairs to form clusters\n",
    "def merge_clusters(clusters):\n",
    "    merged_groups = []\n",
    "    seen = set()\n",
    "    for key, group in clusters.items():\n",
    "        if key not in seen:\n",
    "            merged_group = {key} | set(group)\n",
    "            to_merge = [g for g in merged_groups if g & merged_group]\n",
    "            for g in to_merge:\n",
    "                merged_group |= g\n",
    "                merged_groups.remove(g)\n",
    "            merged_groups.append(merged_group)\n",
    "            seen.update(merged_group)\n",
    "    return merged_groups\n",
    "\n",
    "def form_clusters(candidate_pairs):\n",
    "    clusters_rdd = spark.sparkContext.parallelize(candidate_pairs).groupByKey().mapValues(list).collect()\n",
    "    clusters_dict = defaultdict(set)\n",
    "    for k, v in clusters_rdd:\n",
    "        clusters_dict[k].update(v)\n",
    "        for val in v:\n",
    "            clusters_dict[val].add(k)\n",
    "            clusters_dict[val].update(v)\n",
    "    return merge_clusters(clusters_dict)\n",
    "\n",
    "merged_groups_part1 = form_clusters(filtered_candidates_part1)\n",
    "merged_groups_part2 = form_clusters(filtered_candidates_part2)\n",
    "\n",
    "# Log number of clusters\n",
    "num_clusters_part1 = len(merged_groups_part1)\n",
    "num_clusters_part2 = len(merged_groups_part2)\n",
    "logging.info(f\"Number of clusters for part 1: {num_clusters_part1}\")\n",
    "logging.info(f\"Number of clusters for part 2: {num_clusters_part2}\")\n",
    "\n",
    "# Calculate and log average cluster size\n",
    "average_cluster_size_part1 = sum(len(group) for group in merged_groups_part1) / num_clusters_part1\n",
    "average_cluster_size_part2 = sum(len(group) for group in merged_groups_part2) / num_clusters_part2\n",
    "logging.info(f\"Average cluster size for part 1: {average_cluster_size_part1}\")\n",
    "logging.info(f\"Average cluster size for part 2: {average_cluster_size_part2}\")\n",
    "\n",
    "logging.info(\"Merged similar groups into clusters\")\n",
    "\n",
    "# Function to get the representative process\n",
    "def get_representative_process(group):\n",
    "    min_total_distance = float('inf')\n",
    "    representative_process = None\n",
    "    for pid1 in group:\n",
    "        total_distance = 0\n",
    "        for pid2 in group:\n",
    "            if pid1 != pid2:\n",
    "                total_distance += calculate_edit_distance(pid1, pid2, process_events_broadcast.value)\n",
    "        if total_distance < min_total_distance:\n",
    "            min_total_distance = total_distance\n",
    "            representative_process = pid1\n",
    "    return representative_process\n",
    "\n",
    "# Generate part1Output.txt\n",
    "def generate_part1_output(merged_groups, output_file_path):\n",
    "    with open(output_file_path, 'w') as file:\n",
    "        new_id = 1\n",
    "        for group in merged_groups:\n",
    "            representative_process = get_representative_process(group)\n",
    "            merged_pids = \",\".join(map(str, sorted(group)))\n",
    "            file.write(f\"{new_id}:{merged_pids}\\n\")\n",
    "            _, events = process_events_broadcast.value[representative_process]\n",
    "            for event in events:\n",
    "                file.write(f\"<{event[0]}, {event[1]}, {event[2]}, {event[3]}, {representative_process}>\\n\")\n",
    "            new_id += 1\n",
    "    logging.info(f\"Generated part1Output.txt: {output_file_path}\")\n",
    "\n",
    "generate_part1_output(merged_groups_part1, part1_output_path)\n",
    "\n",
    "# Generate part1Observations.txt\n",
    "def generate_part1_observations(merged_groups, output_file_path):\n",
    "    with open(output_file_path, 'w') as file:\n",
    "        for group in merged_groups:\n",
    "            file.write(f\"Group: {sorted(group)}\\n\")\n",
    "            for pid in sorted(group):\n",
    "                file.write(f\"{pid}:\\n\")\n",
    "                _, events = process_events_dict[pid]\n",
    "                for event in events:\n",
    "                    file.write(f\"<{event[0]}, {event[1]}, {event[2]}, {event[3]}, {pid}>\\n\")\n",
    "            file.write(\"\\n\")\n",
    "    logging.info(f\"Generated part1Observations.txt: {output_file_path}\")\n",
    "\n",
    "generate_part1_observations(merged_groups_part1, part1_observations_path)\n",
    "\n",
    "# Generate part2Observations.txt\n",
    "def generate_part2_observations(merged_groups, output_file_path):\n",
    "    with open(output_file_path, 'w') as file:\n",
    "        for group in merged_groups:\n",
    "            file.write(f\"Group: {sorted(group)}\\n\")\n",
    "            for pid in sorted(group):\n",
    "                file.write(f\"{pid}:\\n\")\n",
    "                _, events = process_events_dict[pid]\n",
    "                for event in events:\n",
    "                    file.write(f\"<{event[0]}, {event[1]}, {event[2]}, {event[3]}, {pid}>\\n\")\n",
    "            file.write(\"\\n\")\n",
    "    logging.info(f\"Generated part2Observations.txt: {output_file_path}\")\n",
    "\n",
    "generate_part2_observations(merged_groups_part2, part2_observations_path)\n",
    "\n",
    "# Close Spark session\n",
    "spark.stop()\n",
    "\n",
    "logging.info(\"Process completed.\")"
   ]
  }
 ],
 "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}