DIS_script-checkpoint.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b34d9af0-8c2d-4c03-a219-53f126b039bc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2024-06-27 15:33:15,864 - INFO - Starting to read the log file\n",
      "2024-06-27 15:33:18,230 - INFO - Completed reading and renaming columns of the log file\n",
      "2024-06-27 15:33:18,264 - INFO - Converted columns to appropriate data types\n",
      "2024-06-27 15:33:18,300 - INFO - Cached the log DataFrame\n",
      "2024-06-27 15:33:18,331 - INFO - Grouped by process_id and collected log entries\n",
      "2024-06-27 15:33:18,365 - INFO - Cached the processes DataFrame\n",
      "2024-06-27 15:33:18,471 - INFO - Transformed features using HashingTF\n",
      "2024-06-27 15:33:18,485 - INFO - Fitted MinHashLSH model\n",
      "2024-06-27 15:33:18,502 - INFO - Transformed data using MinHashLSH model\n",
      "2024-06-27 15:33:18,643 - INFO - Calculated Jaccard similarity and generated candidate pairs\n",
      "2024-06-27 15:36:15,339 - INFO - Filtered candidates for part 2 based on edit distance < 6\n",
      "2024-06-27 15:36:25,760 - INFO - Filtered candidates for part 1 based on edit distance < 3\n",
      "2024-06-27 15:39:20,923 - INFO - Merged similar groups into clusters\n",
      "2024-06-27 15:39:20,925 - INFO - Generated part1Output.txt: C:/Users/dcave/Documents/DIS_project/part1Output.txt\n",
      "2024-06-27 15:39:21,012 - INFO - Generated part1Observations.txt: C:/Users/dcave/Documents/DIS_project/part1Observations.txt\n",
      "2024-06-27 15:39:21,095 - INFO - Generated part2Observations.txt: C:/Users/dcave/Documents/DIS_project/part2Observations.txt\n",
      "2024-06-27 15:39:22,037 - INFO - Process completed.\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, collect_list, trim, array, explode, udf, monotonically_increasing_id\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(100000)\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",
    "# 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 < 6)\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 < 6:\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 < 6\")\n",
    "\n",
    "# Further filter the part 2 candidates for part 1 (edit distance < 3)\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 < 3:\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 < 3\")\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",
    "logging.info(\"Merged similar groups into clusters\")\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",
    "            merged_pids = \",\".join(map(str, group))\n",
    "            file.write(f\"{new_id}:{merged_pids}\\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: {group}\\n\")\n",
    "            for pid in 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: {group}\\n\")\n",
    "            for pid in 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.\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d9f35804-d822-46ae-8a74-95f1073ebee3",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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
}