examples/notebooks/Run_your_first_transform_ray.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Prep Kit - Hello World (Ray)\n",
    "\n",
    "This notebook guides you through running your first data preparation transformation using the data-prep-kit. In this example, we will demonstrate a transformation that takes PDF files as input and extracts their content.\n",
    "\n",
    "[Ray](https://docs.ray.io/en/latest/index.html) is a powerful framework that  enables parallelization while still allowing you to run it efficiently on a local machine, such as your laptop. \n",
    "\n",
    "\n",
    "**Notebook versions:**\n",
    "\n",
    "- Pure python (run locally): [Run_your_first_transform_python.ipynb](Run_your_first_transform_python.ipynb)\n",
    "- Ray version (run locally): this notebook\n",
    "- Google Colab friendly notebook: [Run_your_first_transform_colab.ipynb](Run_your_first_transform_colab.ipynb)  |  [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/IBM/data-prep-kit/blob/dev/examples/notebooks/Run_your_first_transform_colab.ipynb)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step-1: Setting up Python Dev Environment\n",
    "\n",
    "Please follow instructions from [Getting started section](../../README.md#gettingstarted) to setup your python development environment"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step-2: Inspect Data\n",
    "\n",
    "For this example, we will show PDF processing capabilities of DPK.  The PDF documents are in [Input-Test-Data](Input-Test-Data) folder.\n",
    "\n",
    "- [IBM Granite model paper](Input-Test-Data/GraniteCodePaper.pdf) . [Arxiv link](https://arxiv.org/abs/2405.04324)\n",
    "- [Attention is all you need paper](Input-Test-Data/attention_is_all_you_need.pdf)  . [Arxiv link](https://arxiv.org/pdf/1706.03762) - seminal paper on transformer/attention architecture\n",
    "\n",
    "\n",
    "**Try your own PDFs** : You can drop any PDF documents into [Input-Test-Data](Input-Test-Data) folder and run this code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "INPUT_DIR = 'Input-Test-Data'\n",
    "OUTPUT_DIR = 'Output-Test-Data'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step-3: Extract Text from PDF \n",
    "\n",
    "This code is designed to set up a data transformation process that extracts text from PDF.  We will save the output as parquet format."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os, sys\n",
    "import ast\n",
    "\n",
    "# Utilities from the data-prep-kit's data-processing-lib library provide functions and classes for parameter management, \n",
    "from pdf2parquet_transform import (pdf2parquet_contents_type_cli_param, pdf2parquet_contents_types,)\n",
    "from data_processing.utils import GB, ParamsUtils\n",
    "\n",
    "\n",
    "ingest_config = {\n",
    "    pdf2parquet_contents_type_cli_param: pdf2parquet_contents_types.JSON,\n",
    "}\n",
    "\n",
    "# num_cpus_available: Determines the number of CPUs to use for parallel processing. It's set to one-fourth of the total available CPUs on the machine.\n",
    "# worker_options: Specifies the resources each worker will use, including the number of CPUs (num_cpus) and the memory (memory), set to 2 gigabytes (using a utility function GB).\n",
    "\n",
    "num_cpus_available =  os.cpu_count()/4\n",
    "worker_options = {\"num_cpus\" : num_cpus_available, \"memory\": 2 * GB}\n",
    "\n",
    "#local_conf: A dictionary specifying the local input and output folders where the PDF files will be read from and the transformed data will be saved.\n",
    "local_conf = {\n",
    "    \"input_folder\": INPUT_DIR,\n",
    "    \"output_folder\": OUTPUT_DIR,\n",
    "}\n",
    "\n",
    "#params: A dictionary containing various runtime parameters for the transformation.\n",
    "#run_locally: A flag indicating that the transformation should run locally.\n",
    "#data_local_config: Configuration for local data access, such as input and output folders, converted into a format compatible with the transformation using ParamsUtils.convert_to_ast.\n",
    "#data_files_to_use: Specifies that only PDF files (['.pdf']) will be used as input data.\n",
    "#runtime_worker_options: Specifies worker configuration options for Ray, with one gigabyte of memory per worker.\n",
    "#runtime_num_workers: Number of workers to be used for the transformation.\n",
    "#runtime_pipeline_id and runtime_job_id: Identifiers for the pipeline and job, respectively.\n",
    "#runtime_code_location: Provides metadata about the code location, such as its repository and commit details, using the ParamsUtils.convert_to_ast function to format it correctly.\n",
    "\n",
    "params = {\n",
    "    \"run_locally\": True,\n",
    "    \"data_local_config\": ParamsUtils.convert_to_ast(local_conf),\n",
    "    \"data_files_to_use\": ast.literal_eval(\"['.pdf']\"),\n",
    "    \"runtime_worker_options\": ParamsUtils.convert_to_ast(worker_options),\n",
    "    \"runtime_num_workers\": 2,\n",
    "    \"runtime_pipeline_id\": \"pipeline_id\",\n",
    "    \"runtime_job_id\": \"job_id\",\n",
    "}\n",
    "sys.argv = ParamsUtils.dict_to_req(d=(params | ingest_config))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3 - Execute"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now its time to run the transformation.\n",
    "\n",
    "This will launch a *local RAY cluster* to execute our code in parallel (using multiple workers (=2)).  You can view the RAY dashboard in the URL printed below.\n",
    "\n",
    "E.g. http://127.0.0.1:8265 \n",
    "\n",
    "You will notice, that the code will download models to execute the transformation.  These models will be used to process PDFs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "13:28:13 INFO - Running locally\n",
      "13:28:13 INFO - pdf2parquet parameters are : {'artifacts_path': None, 'contents_type': <pdf2parquet_contents_types.JSON: 'application/json'>, 'do_table_structure': True, 'do_ocr': False}\n",
      "13:28:13 INFO - data factory data_ is using local data access: input_folder - Input-Test-Data output_folder - Output-Test-Data\n",
      "13:28:13 INFO - data factory data_ max_files -1, n_sample -1\n",
      "13:28:13 INFO - data factory data_ Not using data sets, checkpointing False, max files -1, random samples -1, files to use ['.pdf'], files to checkpoint ['.parquet']\n",
      "13:28:13 INFO - pipeline id pipeline_id\n",
      "13:28:13 INFO - code location None\n",
      "13:28:13 INFO - number of workers 2 worker options {'num_cpus': 4.0, 'memory': 2147483648, 'max_restarts': -1}\n",
      "13:28:13 INFO - actor creation delay 0\n",
      "13:28:13 INFO - job details {'job category': 'preprocessing', 'job name': 'pdf2parquet', 'job type': 'ray', 'job id': 'job_id'}\n",
      "2024-09-04 13:28:16,713\tINFO worker.py:1744 -- Started a local Ray instance. View the dashboard at \u001b[1m\u001b[32mhttp://127.0.0.1:8265 \u001b[39m\u001b[22m\n",
      "\u001b[36m(orchestrate pid=1191546)\u001b[0m 13:28:21 INFO - orchestrator started at 2024-09-04 13:28:21\n",
      "\u001b[36m(orchestrate pid=1191546)\u001b[0m 13:28:21 INFO - Number of files is 2, source profile {'max_file_size': 2.112621307373047, 'min_file_size': 1.2146415710449219, 'total_file_size': 3.3272628784179688}\n",
      "\u001b[36m(orchestrate pid=1191546)\u001b[0m 13:28:21 INFO - Cluster resources: {'cpus': 16, 'gpus': 1, 'memory': 5.519162750802934, 'object_store': 2.759581374935806}\n",
      "\u001b[36m(orchestrate pid=1191546)\u001b[0m 13:28:21 INFO - Number of workers - 2 with {'num_cpus': 4.0, 'memory': 2147483648, 'max_restarts': -1} each\n",
      "\u001b[36m(orchestrate pid=1191546)\u001b[0m 13:28:21 INFO - Completed 0 files (0.0%)  in 6.210803985595703e-06 min. Waiting for completion\n",
      "\u001b[36m(RayTransformFileProcessor pid=1192507)\u001b[0m 13:28:25 INFO - Initializing models\n",
      "Fetching 7 files: 100%|██████████| 7/7 [00:00<00:00, 58956.08it/s]\n",
      "\u001b[36m(RayTransformFileProcessor pid=1192506)\u001b[0m /home/sujee/apps/anaconda3/envs/data-prep-kit-3-py311/lib/python3.11/site-packages/torch/nn/modules/transformer.py:307: UserWarning: enable_nested_tensor is True, but self.use_nested_tensor is False because encoder_layer.self_attn.batch_first was not True(use batch_first for better inference performance)\n",
      "\u001b[36m(RayTransformFileProcessor pid=1192506)\u001b[0m   warnings.warn(f\"enable_nested_tensor is True, but self.use_nested_tensor is False because {why_not_sparsity_fast_path}\")\n",
      "\u001b[36m(orchestrate pid=1191546)\u001b[0m 13:29:32 INFO - Completed processing 2 files in 1.1829445679982504 min\n",
      "\u001b[36m(orchestrate pid=1191546)\u001b[0m 13:29:32 INFO - done flushing in 0.0008149147033691406 sec\n",
      "\u001b[36m(RayTransformFileProcessor pid=1192506)\u001b[0m 13:28:25 INFO - Initializing models\n",
      "Fetching 7 files: 100%|██████████| 7/7 [00:00<00:00, 164023.06it/s]\n",
      "\u001b[36m(RayTransformFileProcessor pid=1192507)\u001b[0m /home/sujee/apps/anaconda3/envs/data-prep-kit-3-py311/lib/python3.11/site-packages/torch/nn/modules/transformer.py:307: UserWarning: enable_nested_tensor is True, but self.use_nested_tensor is False because encoder_layer.self_attn.batch_first was not True(use batch_first for better inference performance)\n",
      "\u001b[36m(RayTransformFileProcessor pid=1192507)\u001b[0m   warnings.warn(f\"enable_nested_tensor is True, but self.use_nested_tensor is False because {why_not_sparsity_fast_path}\")\n",
      "13:29:42 INFO - Completed execution in 1.479382046063741 min, execution result 0\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "✅ Tranformation run completed successfully\n",
      "CPU times: user 485 ms, sys: 281 ms, total: 766 ms\n",
      "Wall time: 1min 30s\n"
     ]
    }
   ],
   "source": [
    "%%time \n",
    "\n",
    "from data_processing_ray.runtime.ray import RayTransformLauncher\n",
    "from pdf2parquet_transform_ray import Pdf2ParquetRayTransformConfiguration\n",
    "\n",
    "launcher = RayTransformLauncher(Pdf2ParquetRayTransformConfiguration())\n",
    "return_code = launcher.launch()\n",
    "\n",
    "if return_code == 0:\n",
    "    print (f\"✅ Tranformation run completed successfully\")\n",
    "else:\n",
    "    raise Exception (\"❌ Transformation run failed\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step-4: Inspect the generated output\n",
    "\n",
    "We will use pandas to read parquet files and display.\n",
    "\n",
    "You should see one-entry per PDF input file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import glob\n",
    "import pandas as pd\n",
    "\n",
    "## Reads parquet files in a folder into a pandas dataframe \n",
    "def read_parquet_files_as_df (parquet_dir):\n",
    "    parquet_files = glob.glob(f'{parquet_dir}/*.parquet')\n",
    "\n",
    "    # read each parquet file into a DataFrame and store in a list\n",
    "    dfs = [pd.read_parquet (f) for f in parquet_files]\n",
    "\n",
    "    # Concatenate all DataFrames into a single DataFrame\n",
    "    data_df = pd.concat(dfs, ignore_index=True)\n",
    "    return data_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>filename</th>\n",
       "      <th>contents</th>\n",
       "      <th>num_pages</th>\n",
       "      <th>num_tables</th>\n",
       "      <th>num_doc_elements</th>\n",
       "      <th>document_id</th>\n",
       "      <th>ext</th>\n",
       "      <th>hash</th>\n",
       "      <th>size</th>\n",
       "      <th>date_acquired</th>\n",
       "      <th>pdf_convert_time</th>\n",
       "      <th>source_filename</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>GraniteCodePaper.pdf</td>\n",
       "      <td>{\"_name\":\"\",\"type\":\"pdf-document\",\"description...</td>\n",
       "      <td>28</td>\n",
       "      <td>17</td>\n",
       "      <td>320</td>\n",
       "      <td>6769e793-6e20-4dfc-8fe6-12ecc1944aac</td>\n",
       "      <td>pdf</td>\n",
       "      <td>d1f30543fce4e18d223cf4ba7728733e169201b4df5d5d...</td>\n",
       "      <td>584819</td>\n",
       "      <td>2024-09-04T13:29:32.380832</td>\n",
       "      <td>53.872901</td>\n",
       "      <td>GraniteCodePaper.pdf</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>attention_is_all_you_need.pdf</td>\n",
       "      <td>{\"_name\":\"\",\"type\":\"pdf-document\",\"description...</td>\n",
       "      <td>15</td>\n",
       "      <td>4</td>\n",
       "      <td>193</td>\n",
       "      <td>20925bb2-4e14-4e58-8ae1-0ce971dfc2bf</td>\n",
       "      <td>pdf</td>\n",
       "      <td>fee309974aabb59c48dbfaeb011ee8a2c78f2e492747b9...</td>\n",
       "      <td>131167</td>\n",
       "      <td>2024-09-04T13:28:57.169495</td>\n",
       "      <td>19.068189</td>\n",
       "      <td>attention_is_all_you_need.pdf</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                        filename  \\\n",
       "0           GraniteCodePaper.pdf   \n",
       "1  attention_is_all_you_need.pdf   \n",
       "\n",
       "                                            contents  num_pages  num_tables  \\\n",
       "0  {\"_name\":\"\",\"type\":\"pdf-document\",\"description...         28          17   \n",
       "1  {\"_name\":\"\",\"type\":\"pdf-document\",\"description...         15           4   \n",
       "\n",
       "   num_doc_elements                           document_id  ext  \\\n",
       "0               320  6769e793-6e20-4dfc-8fe6-12ecc1944aac  pdf   \n",
       "1               193  20925bb2-4e14-4e58-8ae1-0ce971dfc2bf  pdf   \n",
       "\n",
       "                                                hash    size  \\\n",
       "0  d1f30543fce4e18d223cf4ba7728733e169201b4df5d5d...  584819   \n",
       "1  fee309974aabb59c48dbfaeb011ee8a2c78f2e492747b9...  131167   \n",
       "\n",
       "                date_acquired  pdf_convert_time                source_filename  \n",
       "0  2024-09-04T13:29:32.380832         53.872901           GraniteCodePaper.pdf  \n",
       "1  2024-09-04T13:28:57.169495         19.068189  attention_is_all_you_need.pdf  "
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "output_df = read_parquet_files_as_df(OUTPUT_DIR)\n",
    "output_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('[\\n'\n",
      " '    '\n",
      " '\"{\\\\\"_name\\\\\":\\\\\"\\\\\",\\\\\"type\\\\\":\\\\\"pdf-document\\\\\",\\\\\"description\\\\\":{\\\\\"logs\\\\\":[]},\\\\\"file-info\\\\\":{\\\\\"filename\\\\\":\\\\\"GraniteCodePaper.pdf\\\\\",\\\\\"document-hash\\\\\":\\\\\"cd6f480d8c03baf8fc4fc5a874dfa726764028fd6fb032532b6807522da6df48\\\\\",\\\\\"#-pages\\\\\":28,\\\\\"page-hashes\\\\\":[{\\\\\"hash\\\\\":\\\\\"2683abedc4e9b0a7ecffbba4e696db4cdeeaaa770de105f14b46cf7d0fef935a\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":1},{\\\\\"hash\\\\\":\\\\\"6b67e3f7d141634b469b81edae327289f954805104902cb51eced075585ecdf5\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":2},{\\\\\"hash\\\\\":\\\\\"d6c88ae80a41c48e99bacf83121051d01010c0edd5489fbc1c34b1f11f7d45c8\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":3},{\\\\\"hash\\\\\":\\\\\"a342fd3b69659664f36d83fba38be16f7a78bd8fa31cc3216c884a856c42c842\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":4},{\\\\\"hash\\\\\":\\\\\"bd8ab5e38ea21b04545dd2c9a73cd82d17e6e732f497125aca4e9d8468891e7d\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":5},{\\\\\"hash\\\\\":\\\\\"b24778e5dd56e510671b10b05e099b3ce72178a6a49cc6241dcf52ac1aaea27a\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":6},{\\\\\"hash\\\\\":\\\\\"3a279bd1e60253b7dfc02e5644d353aa9f98fd0a94b45d2dbabc616f526fa494\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":7},{\\\\\"hash\\\\\":\\\\\"3f333f5b7d034c7933e125a454eb419f6852ed085ba3fc5b2d43bfadb590cc33\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":8},{\\\\\"hash\\\\\":\\\\\"4baa552c46312b9718e4373d7ddd4de85b018bb5cd07aca51ab6034cadf5103c\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":9},{\\\\\"hash\\\\\":\\\\\"7fd102f1856ae7c94c9bdd41bc1d39092cde1161eb6ddeaf5f2583be611af981\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":10},{\\\\\"hash\\\\\":\\\\\"802a0f5ad7a141042cf8cb9f9ae9d0b5b596eba4079f86caef4ea31e51dfb8b3\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":11},{\\\\\"hash\\\\\":\\\\\"d53f72a373976815a57835da8d1c39d186599722fe6ac8e9ecca9433ecbea04f\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":12},{\\\\\"hash\\\\\":\\\\\"0ada0a41017360f82e16e77a6f3d9f7e1a9602514d964878aeaf0cf456afe3af\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":13},{\\\\\"hash\\\\\":\\\\\"d2aaf56f2d6637e58ca064bce92b30a96602a328c858badfc0f95e432775a180\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":14},{\\\\\"hash\\\\\":\\\\\"541a95ae9bc0caf6ad61fed7646891662545bece1c42ee213d65ef726ccb3707\\\\\",\\\\\"model\\\\\":\\\\\"default\\\\\",\\\\\"page\\\\\":15},{\\\\\"hash\\\\\":\\\\\"eba5296396cb609e3cc72d8bc09bbf746e8b')\n"
     ]
    }
   ],
   "source": [
    "# Inspect contents\n",
    "\n",
    "import json\n",
    "import pprint\n",
    "\n",
    "column_list = output_df['contents'].tolist()\n",
    "column_json = json.dumps(column_list, indent=4)\n",
    "pprint.pprint(column_json[:2000]) # display first few lines"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "data-prep-kit-3-py311",
   "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.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}