notebooks/1.0_data_prep.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "79b5556f-5146-4807-9dfe-65372412c9b9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# clean up the data\n",
    "!sed -n -r '/[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}:[^\\s]+/p' data/combined.txt > data/comb_out.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a554e522-3317-427b-9ddf-94b645add1c0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# convert to csv\n",
    "!sed 's/:/,/' data/comb_out.txt > data/comb_out.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "d7bc1386-39ad-4d92-865c-855bf8bc1fe0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2687437913 data/comb_out.csv\n"
     ]
    }
   ],
   "source": [
    "# total count\n",
    "!wc -l data/comb_out.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ff8563df-0032-44fb-83fc-5eaa046d28b9",
   "metadata": {},
   "outputs": [],
   "source": [
    "!python3 reservoir_sampling.py data/comb_out.csv 1000000 > data/comb_1m.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c4cf810f-04db-4f53-bb09-0ccc36150ec9",
   "metadata": {},
   "outputs": [],
   "source": [
    "## reservoir_sampling.py\n",
    "import sys\n",
    "import random\n",
    "\n",
    "filename = sys.argv[1]\n",
    "sample_size = int(sys.argv[2])\n",
    "reservoir = []\n",
    "\n",
    "with open(filename, 'r') as f:\n",
    "    for i, line in enumerate(f):\n",
    "        if i < sample_size:\n",
    "            reservoir.append(line)\n",
    "        else:\n",
    "            r = random.randint(0, i)\n",
    "            if r < sample_size:\n",
    "                reservoir[r] = line\n",
    "\n",
    "for line in reservoir:\n",
    "    print(line, end='')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4cfed27b-462e-467b-aac9-3ff102bc0a6d",
   "metadata": {},
   "outputs": [],
   "source": [
    "!sed 's/,\\(.*$\\)/,\"\\1\"/g' comb_1m.csv > comb_1m_out.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6eb16660-05a1-48df-9c3e-16c560b1cc25",
   "metadata": {},
   "source": [
    "Average record size = Total file size / Number of records <br>\n",
    "                      = 82 GB / 2,687,437,913 records <br>\n",
    "                      = approximately 31.5 bytes/record <br>\n",
    "\n",
    "Number of records for 3 GB ≈ 3 GB / average record size <br>\n",
    "                           ≈ (3 * 1024 * 1024 * 1024) bytes / 31.5 bytes/record <br>\n",
    "                           ≈ 102,236,404 records <br>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "1b8f916e-9778-4e08-819a-e3226fbade4a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import multiprocessing\n",
    "import random\n",
    "import os\n",
    "\n",
    "# Define the number of records to sample\n",
    "num_records_to_sample = 102236404 * 3\n",
    "total_records = 2687437913\n",
    "record_size = 31.5  # This should be adjusted to your actual average record size\n",
    "\n",
    "# Calculate number of records per chunk for parallel processing\n",
    "num_processes = multiprocessing.cpu_count()  # Number of processes to create\n",
    "chunk_size = total_records // num_processes\n",
    "\n",
    "# Function to process each chunk and select records\n",
    "def process_chunk(chunk_index, return_dict):\n",
    "    # Calculate the start and end index of the chunk\n",
    "    start_index = chunk_index * chunk_size\n",
    "    end_index = start_index + chunk_size if chunk_index < num_processes - 1 else total_records\n",
    "    chunk_records_to_sample = num_records_to_sample // num_processes\n",
    "\n",
    "    selected_records = set(random.sample(range(start_index, end_index), chunk_records_to_sample))\n",
    "\n",
    "    selected_data = []\n",
    "    with open('data/comb_out.csv', 'r') as file:\n",
    "        file.seek(int(start_index * record_size))  # Seek to the start of the chunk\n",
    "        for i in range(start_index, end_index):\n",
    "            record = file.readline()\n",
    "            if i in selected_records:\n",
    "                selected_data.append(record)\n",
    "            if not record:  # End of file\n",
    "                break\n",
    "\n",
    "    # Store result in return dictionary\n",
    "    return_dict[chunk_index] = selected_data\n",
    "\n",
    "# Create a manager to hold the returned data from each process\n",
    "manager = multiprocessing.Manager()\n",
    "return_dict = manager.dict()\n",
    "\n",
    "# Create a list to keep track of processes\n",
    "processes = []\n",
    "\n",
    "# Create and start processes\n",
    "for i in range(num_processes):\n",
    "    p = multiprocessing.Process(target=process_chunk, args=(i, return_dict))\n",
    "    processes.append(p)\n",
    "    p.start()\n",
    "\n",
    "# Ensure all processes have finished execution\n",
    "for p in processes:\n",
    "    p.join()\n",
    "\n",
    "# Combine the data from all processes\n",
    "combined_data = []\n",
    "for chunk_data in return_dict.values():\n",
    "    combined_data.extend(chunk_data)\n",
    "\n",
    "# Now `combined_data` holds the randomly selected records\n",
    "\n",
    "# Write the combined data to the output file\n",
    "with open('data/comb_9g.csv', 'w') as outfile:\n",
    "    outfile.writelines(combined_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "77c7b02a-bba9-4ac6-a49f-001dc65ab35c",
   "metadata": {},
   "outputs": [],
   "source": [
    "!sed 's/,\\(.*$\\)/,\"\\1\"/g' data/comb_9g.csv > data/comb_9g_out.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7e63de24-c40d-46c4-b3eb-efd189b8786b",
   "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.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}