{
 "cells": [
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "---\n",
    "sidebar_label: \"csv-to-avro-or-parquet\"\n",
    "sidebar_position: 2\n",
    "---"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Convert CSV To Parquet Or Avro\n",
    "\n",
    "\n",
    "[![stars - badge-generator](https://img.shields.io/github/stars/bacalhau-project/bacalhau?style=social)](https://github.com/bacalhau-project/bacalhau)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "XIRH0Cr0HOkd"
   },
   "source": [
    "## Introduction"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "cRCGwu83HV3_"
   },
   "source": [
    "Converting from csv to parquet or avro reduces the size of file and allows for faster read and write speeds. With Bacalhau, you can convert your csv files stored on ipfs or on the web without the need to download files and install dependencies locally.\n",
    "\n",
    "In this example tutorial we will convert a csv file from a url to parquet format and save the converted parquet file to IPFS\n",
    "\n",
    "## TD;LR\n",
    "Converting CSV stored in public storage with Bacalhau\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prerequisites\n",
    "\n",
    "To get started, you need to install the Bacalhau client, see more information [here](https://docs.bacalhau.org/getting-started/installation)"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "!command -v bacalhau >/dev/null 2>&1 || (export BACALHAU_INSTALL_DIR=.; curl -sL https://get.bacalhau.org/install.sh | bash)\n",
    "path=!echo $PATH\n",
    "%env PATH=./:{path[0]}"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "QoP0OtYeHYLU"
   },
   "source": [
    "## Running CSV to Arvo or Parquet Locally​\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "X0tYF0CiHcAc"
   },
   "source": [
    "Installing dependencies\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "SI0WIiXkHglL",
    "outputId": "e0ffcdca-5abb-49ee-bc41-4693ac6a3eaf",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [],
   "source": [
    "%cd csv_to_avro_or_parquet"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "5t90v6OkJd5j"
   },
   "source": [
    "## Install the following dependencies \n",
    "\n",
    "Run the following commands:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "%%bash\n",
    "pip install fastavro"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "%%bash\n",
    "pip install numpy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "%%bash\n",
    "pip install pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "pip install pyarrow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "id": "JJnBqegLJxiy",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "python3 src/converter.py ./movies.csv  ./movies.parquet parquet\n",
    "\n",
    "# python converter.py path_to_csv path_to_result_file extension"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "tSZy_yweJ9ii"
   },
   "source": [
    "Viewing the parquet file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 990
    },
    "id": "C7Wb6dU0J3Ha",
    "outputId": "e99f0a1a-5361-40e1-e560-45d55d7b1c25",
    "tags": []
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "pd.read_parquet('./movies.parquet').head()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "WYvBBuatKeQC"
   },
   "source": [
    "## Containerize Script with Docker\n",
    "\n",
    ":::info\n",
    "You can skip this section entirely and directly go to running on bacalhau\n",
    ":::\n",
    "\n",
    "To build your own docker container, create a `Dockerfile`, which contains instructions to build your image.\n",
    "\n",
    "```\n",
    "FROM python:3.8\n",
    "\n",
    "RUN apt update && apt install git\n",
    "\n",
    "RUN git clone https://github.com/bacalhau-project/Sparkov_Data_Generation\n",
    "\n",
    "WORKDIR /Sparkov_Data_Generation/\n",
    "\n",
    "RUN pip3 install -r requirements.txt\n",
    "```\n",
    "\n",
    ":::info\n",
    "See more information on how to containerize your script/app[here](https://docs.docker.com/get-started/02_our_app/)\n",
    ":::\n",
    "\n",
    "\n",
    "### Build the container\n",
    "\n",
    "We will run `docker build` command to build the container;\n",
    "\n",
    "```\n",
    "docker build -t <hub-user>/<repo-name>:<tag> .\n",
    "```\n",
    "\n",
    "Before running the command replace;\n",
    "\n",
    "- **hub-user** with your docker hub username, If you don’t have a docker hub account [follow these instructions to create docker account](https://docs.docker.com/docker-id/), and use the username of the account you created\n",
    "\n",
    "- **repo-name** with the name of the container, you can name it anything you want\n",
    "\n",
    "- **tag** this is not required but you can use the latest tag\n",
    "\n",
    "In our case:\n",
    "\n",
    "```\n",
    "docker build -t jsacex/csv-to-arrow-or-parquet\n",
    "```\n",
    "\n",
    "### Push the container\n",
    "\n",
    "Next, upload the image to the registry. This can be done by using the Docker hub username, repo name or tag.\n",
    "\n",
    "```\n",
    "docker push <hub-user>/<repo-name>:<tag>\n",
    "```\n",
    "\n",
    "In our case:\n",
    "\n",
    "```\n",
    "docker push jsacex/csv-to-arrow-or-parquet\n",
    "```"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "HpQH-lqBRkdT"
   },
   "source": [
    "## Running a Bacalhau Job\n",
    "\n",
    "To submit a job, we are going to either mount the script from a IPFS or from an URL."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "Dwm4cEwqT9nh"
   },
   "source": [
    "### Mounting the CSV File from IPFS\n",
    "\n",
    "With the command below, we are gmounting the CSV file for transactions from IPFS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!command -v bacalhau >/dev/null 2>&1 || (export BACALHAU_INSTALL_DIR=.; curl -sL https://get.bacalhau.org/install.sh | bash)\n",
    "path=!echo $PATH\n",
    "%env PATH=./:{path[0]}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "id": "9OqkBR5fTf1L",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [],
   "source": [
    "%%bash --out job_id\n",
    "bacalhau docker run \\\n",
    "-i ipfs://QmTAQMGiSv9xocaB4PUCT5nSBHrf9HZrYj21BAZ5nMTY2W  \\\n",
    "--wait \\\n",
    "--id-only \\\n",
    "jsacex/csv-to-arrow-or-parquet \\\n",
    "-- python3 src/converter.py ../inputs/transactions.csv  ../outputs/transactions.parquet parquet"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Structure of the command\n",
    "\n",
    "Let's look closely at the command above:\n",
    "\n",
    "* `bacalhau docker run`: call to bacalhau \n",
    "  \n",
    "* `-i ipfs://QmTAQMGiSv9xocaB4PUCT5nSBHrf9HZrYj21BAZ5nMTY2W`: CIDs to use on the job. Mounts them at '/inputs' in the execution.\n",
    "\n",
    "* `jsacex/csv-to-arrow-or-parque`: the name and the tag of the docker image we are using\n",
    "\n",
    "* `../inputs/movies.csv `: path to input dataset\n",
    "\n",
    "* `../outputs/movies.parquet parquet`: path to output\n",
    "\n",
    "* `python3 src/converter.py`: execute the script"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "f0AGj0tQUviw"
   },
   "source": [
    "### Mounting the CSV File from a URL\n",
    "To mount the CSV file from a URL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash --out job_id\n",
    "bacalhau docker run \\\n",
    "-i https://raw.githubusercontent.com/bacalhau-project/csv_to_avro_or_parquet/master/movies.csv \\\n",
    "jsacex/csv-to-arrow-or-parquet \\\n",
    "-- python3 src/converter.py ../inputs/movies.csv  ../outputs/movies.parquet parquet"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Structure of the command\n",
    "\n",
    "Let's look closely at the command above:\n",
    "\n",
    "* `bacalhau docker run`: call to bacalhau \n",
    "  \n",
    "* `-i https://raw.githubusercontent.com/bacalhau-project/csv_to_avro_or_parquet/master/movies.csv`: URL:path of the input data volumes downloaded from a URL source\n",
    "\n",
    "* `jsacex/csv-to-arrow-or-parque`: the name and the tag of the docker image we are using\n",
    "\n",
    "* `../inputs/movies.csv `: path to input dataset\n",
    "\n",
    "* `../outputs/movies.parquet parquet`: path to output\n",
    "\n",
    "* `python3 src/converter.py`: execute the script"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When a job is submitted, Bacalhau prints out the related `job_id`. We store that in an environment variable so that we can reuse it later on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "iWfPxY9HMkmX",
    "outputId": "647eac06-eeae-4c5e-d9d3-04459f9cce5b",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "env: JOB_ID=bacalhau describe 71ecde0e-dac3-4c8d-bf2e-7a92cc54425e\n"
     ]
    }
   ],
   "source": [
    "%env JOB_ID={job_id}"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Checking the State of your Jobs\n",
    "\n",
    "- **Job status**: You can check the status of the job using `bacalhau list`. \n",
    "\n",
    ":::note\n",
    "Replace the {JOB_ID} with your generated ID.\n",
    ":::"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "dErDkTWoMkmX",
    "outputId": "0ca2c81e-9632-4b71-a737-2cb1c5bfc5bc",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[92;100m CREATED  \u001b[0m\u001b[92;100m ID       \u001b[0m\u001b[92;100m JOB                     \u001b[0m\u001b[92;100m STATE     \u001b[0m\u001b[92;100m VERIFIED \u001b[0m\u001b[92;100m PUBLISHED               \u001b[0m\n",
      "\u001b[97;40m 16:53:30 \u001b[0m\u001b[97;40m 71ecde0e \u001b[0m\u001b[97;40m Docker jsacex/csv-to... \u001b[0m\u001b[97;40m Completed \u001b[0m\u001b[97;40m          \u001b[0m\u001b[97;40m ipfs://QmP5PbbJZ1fdq... \u001b[0m\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "bacalhau list --id-filter={JOB_ID} "
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "kFYpNA32c7t5"
   },
   "source": [
    "\n",
    "When it says `Published` or `Completed`, that means the job is done, and we can get the results.\n",
    "\n",
    "- **Job information**: You can find out more information about your job by using `bacalhau describe`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "z7wlx10bMkmY",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "bacalhau describe {JOB_ID}"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "2I4DHnt0Vzua"
   },
   "source": [
    "- **Job download**: You can download your job results directly by using `bacalhau get`. Alternatively, you can choose to create a directory to store your results. In the command below, we created a directory and downloaded our job output to be stored in that directory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "9-uJ7_F7MkmY",
    "outputId": "3b1c762e-1d77-491d-b9ab-17d6b27ba6af",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "rm -rf results && mkdir -p results\n",
    "bacalhau get ${JOB_ID} --output-dir results"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "nrZcHon2WOd-"
   },
   "source": [
    "## Viewing your Job Output\n",
    "\n",
    "To view the file, run the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "m_tquvEGMkmZ",
    "outputId": "c82d6070-10bd-460c-ce61-31bba1d989ba",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "ls results/outputs"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "id": "7J2UmPZGWwog"
   },
   "source": [
    "Alternatively, you can do this."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 239
    },
    "id": "rkECQ7Y6Wvne",
    "outputId": "666f8c7f-e56a-44cb-cd34-46e2261c1a8e",
    "tags": [
     "skip-execution"
    ]
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import os\n",
    "pd.read_parquet('results/outputs/transactions.parquet')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Need Support?\n",
    "\n",
    "For questions, feedback, please reach out in our [forum](https://github.com/filecoin-project/bacalhau/discussions)"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "provenance": []
  },
  "kernelspec": {
   "display_name": "Python 3.10.6 64-bit",
   "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.1"
  },
  "vscode": {
   "interpreter": {
    "hash": "916dbcbb3f70747c44a77c7bcd40155683ae19c65e1c03b4aa3499c5328201f1"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}