examples/notebooks/Run_your_first_transform.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "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. We will show how to execute this process using Ray to enable parallelization while still allowing you to run it efficiently on a local machine, such as your laptop. \n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#The code is designed to set up a data transformation process that converts PDF files to a different format (like Parquet) \n",
    "# using a parallelized approach. The transformation can run both locally and in a distributed environment using Ray, a framework for \n",
    "# parallel and distributed computing.\n",
    "\n",
    "#Import necessary libraries\n",
    "import os, sys\n",
    "import shutil\n",
    "import ast\n",
    "\n",
    "# Utilities from the data-prep-kit's data-processing-lib library provide functions and classes for parameter management, \n",
    "# memory calculations, and launchers for running transformations either locally or using Ray.\n",
    "from pdf2parquet_transform_python import Pdf2ParquetPythonTransformConfiguration\n",
    "from pdf2parquet_transform_ray import Pdf2ParquetRayTransformConfiguration\n",
    "from pdf2parquet_transform import (pdf2parquet_contents_type_cli_param, pdf2parquet_contents_types,)\n",
    "from data_processing.utils import GB, ParamsUtils\n",
    "from data_processing_ray.runtime.ray import RayTransformLauncher\n",
    "from data_processing.runtime.pure_python import PythonTransformLauncher\n",
    "\n",
    "\n",
    "#Set configuration parameters to run. These will be used to parallelize your code. \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",
    "#code_location: Contains metadata about the code location, such as the GitHub repository, commit hash, and file path.\n",
    "#ingest_config: Defines the input data type for the transformation, set to JSON using parameters from the pdf2parquet_transform module.\n",
    "    \n",
    "num_cpus_available =  os.cpu_count()/4\n",
    "worker_options = {\"num_cpus\" : num_cpus_available, \"memory\": 2 * GB}\n",
    "code_location = {\"github\": \"NA\", \"commit_hash\": \"0\", \"path\": \"path\"}\n",
    "ingest_config = {\n",
    "    pdf2parquet_contents_type_cli_param: pdf2parquet_contents_types.JSON,\n",
    "}\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-Test-Data\",\n",
    "    \"output_folder\": \"Output-Test-Data\",\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 half a 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({\"num_cpus\" : num_cpus_available, \"memory\": .5 * GB}),\n",
    "    \"runtime_num_workers\": 2,\n",
    "    \"runtime_pipeline_id\": \"pipeline_id\",\n",
    "    \"runtime_job_id\": \"job_id\",\n",
    "    \"runtime_code_location\": ParamsUtils.convert_to_ast(code_location),\n",
    "}\n",
    "sys.argv = ParamsUtils.dict_to_req(d=(params | ingest_config))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The next code cell is the execution step where the transformation process is actually launched using Ray, a parallel computing framework. RayTransformLauncher is a class for running data transformation jobs using Ray, and Pdf2ParquetRayTransformConfiguration defines the necessary settings for converting PDFs to Parquet format. The launcher instance combines these to execute the transformation with the specified configuration."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "launcher = RayTransformLauncher(Pdf2ParquetRayTransformConfiguration())\n",
    "return_code = launcher.launch()\n",
    "\n",
    "if return_code != 0:\n",
    "    print (f\"Computation failed\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Inspect the output\n",
    "\n",
    "import pandas as pd\n",
    "import json\n",
    "import pprint\n",
    "df = pd.read_parquet(\"Output-Test-Data/GraniteCodePaper.parquet\")\n",
    "column_list = df['contents'].tolist()\n",
    "column_json = json.dumps(column_list, indent=6)\n",
    "print(column_json)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "dpk",
   "language": "python",
   "name": "data-prep-kit"
  },
  "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
}