Convert TSV Data To Parquet.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convert TSV Data To Parquet\n",
    "\n",
    "It's simple to convert TSV data into [Apache Parquet](https://parquet.apache.org/) file format."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import sagemaker\n",
    "\n",
    "# Get region \n",
    "session = boto3.session.Session()\n",
    "region_name = session.region_name\n",
    "\n",
    "# Get SageMaker session & default S3 bucket\n",
    "sagemaker_session = sagemaker.Session()\n",
    "bucket = sagemaker_session.default_bucket()\n",
    "\n",
    "# Set S3 prefixes\n",
    "parquet_prefix = 'amazon-reviews-pds/parquet'\n",
    "\n",
    "# Set S3 path to Parquet data\n",
    "s3_path_parquet = 's3://{}/{}'.format(bucket, parquet_prefix)\n",
    "\n",
    "# Set Athena parameters\n",
    "database_name = 'coeaws'\n",
    "table_name_tsv = 'amazon_reviews_tsv'\n",
    "table_name_parquet = 'amazon_reviews_parquet'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# install pyathena\n",
    "!pip install -q PyAthena==1.8.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyathena import connect\n",
    "from pyathena.pandas_cursor import PandasCursor\n",
    "from pyathena.util import as_pandas"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Parquet Files from TSV Table\n",
    "\n",
    "As you can see from the query below, we’re also adding a new `year` column to our dataset by converting the `review_date` string to a date format, and then cast the year out of the date. Let’s store the year value as an integer. And let's partition the Parquet data by `Product Category`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set S3 staging directory -- this is a temporary directory used for Athena queries\n",
    "s3_staging_dir = 's3://{0}/athena/staging'.format(bucket)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CREATE TABLE IF NOT EXISTS coeaws.amazon_reviews_parquet\n",
      "WITH (format = 'PARQUET', external_location = 's3://sagemaker-us-east-2-533787958253/amazon-reviews-pds/parquet', partitioned_by = ARRAY['product_category']) AS\n",
      "SELECT marketplace,\n",
      "         customer_id,\n",
      "         review_id,\n",
      "         product_id,\n",
      "         product_parent,\n",
      "         product_title,\n",
      "         star_rating,\n",
      "         helpful_votes,\n",
      "         total_votes,\n",
      "         vine,\n",
      "         verified_purchase,\n",
      "         review_headline,\n",
      "         review_body,\n",
      "         CAST(YEAR(DATE(review_date)) AS INTEGER) AS year,\n",
      "         DATE(review_date) AS review_date,\n",
      "         product_category\n",
      "FROM coeaws.amazon_reviews_tsv\n"
     ]
    }
   ],
   "source": [
    "# SQL statement to execute\n",
    "statement = \"\"\"CREATE TABLE IF NOT EXISTS {}.{}\n",
    "WITH (format = 'PARQUET', external_location = '{}', partitioned_by = ARRAY['product_category']) AS\n",
    "SELECT marketplace,\n",
    "         customer_id,\n",
    "         review_id,\n",
    "         product_id,\n",
    "         product_parent,\n",
    "         product_title,\n",
    "         star_rating,\n",
    "         helpful_votes,\n",
    "         total_votes,\n",
    "         vine,\n",
    "         verified_purchase,\n",
    "         review_headline,\n",
    "         review_body,\n",
    "         CAST(YEAR(DATE(review_date)) AS INTEGER) AS year,\n",
    "         DATE(review_date) AS review_date,\n",
    "         product_category\n",
    "FROM {}.{}\"\"\".format(database_name, table_name_parquet, s3_path_parquet, database_name, table_name_tsv)\n",
    "\n",
    "print(statement)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Execute statement using connection cursor\n",
    "This can take a few minutes.  Please be patient."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyathena.cursor.Cursor at 0x7f67c1e305c0>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cursor = connect(region_name=region_name, s3_staging_dir=s3_staging_dir).cursor()\n",
    "cursor.execute(statement)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load partitions by running `MSCK REPAIR TABLE`\n",
    "to load the Parquet partitions. To do so, just issue the following SQL command: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "MSCK REPAIR TABLE coeaws.amazon_reviews_parquet\n"
     ]
    }
   ],
   "source": [
    "statement = 'MSCK REPAIR TABLE {}.{}'.format(database_name, table_name_parquet)\n",
    "\n",
    "print(statement)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyathena.cursor.Cursor at 0x7f67c1b68668>"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cursor = connect(region_name=region_name, s3_staging_dir=s3_staging_dir).cursor()\n",
    "cursor.execute(statement)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SHOW PARTITIONS coeaws.amazon_reviews_parquet\n"
     ]
    }
   ],
   "source": [
    "# Show the partitions\n",
    "statement = 'SHOW PARTITIONS {}.{}'.format(database_name, table_name_parquet)\n",
    "\n",
    "print(statement)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "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>partition</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>product_category=Digital_Software</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>product_category=Digital_Video_Games</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                              partition\n",
       "0     product_category=Digital_Software\n",
       "1  product_category=Digital_Video_Games"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cursor = connect(region_name=region_name, s3_staging_dir=s3_staging_dir).cursor()\n",
    "cursor.execute(statement)\n",
    "\n",
    "df_partitions = as_pandas(cursor)\n",
    "df_partitions.head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SELECT * FROM coeaws.amazon_reviews_parquet\n",
      "    WHERE product_category = 'Digital_Software' LIMIT 100\n"
     ]
    }
   ],
   "source": [
    "# Run a sample query\n",
    "product_category = 'Digital_Software'\n",
    "\n",
    "statement = \"\"\"SELECT * FROM {}.{}\n",
    "    WHERE product_category = '{}' LIMIT 100\"\"\".format(database_name, table_name_parquet, product_category)\n",
    "\n",
    "print(statement)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "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>marketplace</th>\n",
       "      <th>customer_id</th>\n",
       "      <th>review_id</th>\n",
       "      <th>product_id</th>\n",
       "      <th>product_parent</th>\n",
       "      <th>product_title</th>\n",
       "      <th>star_rating</th>\n",
       "      <th>helpful_votes</th>\n",
       "      <th>total_votes</th>\n",
       "      <th>vine</th>\n",
       "      <th>verified_purchase</th>\n",
       "      <th>review_headline</th>\n",
       "      <th>review_body</th>\n",
       "      <th>year</th>\n",
       "      <th>review_date</th>\n",
       "      <th>product_category</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>US</td>\n",
       "      <td>18640556</td>\n",
       "      <td>RWKGA4N6R4KMY</td>\n",
       "      <td>B00C1QCBY4</td>\n",
       "      <td>640306152</td>\n",
       "      <td>Sketchbook Pro 6</td>\n",
       "      <td>5</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>N</td>\n",
       "      <td>Y</td>\n",
       "      <td>Oustanding</td>\n",
       "      <td>It's a much simpler application than Painter o...</td>\n",
       "      <td>2014</td>\n",
       "      <td>2014-05-14</td>\n",
       "      <td>Digital_Software</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>US</td>\n",
       "      <td>11866626</td>\n",
       "      <td>RF8V3GF4MGT6M</td>\n",
       "      <td>B00H9A60O4</td>\n",
       "      <td>608720080</td>\n",
       "      <td>Avast Free Antivirus 2015 [Download]</td>\n",
       "      <td>5</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>N</td>\n",
       "      <td>N</td>\n",
       "      <td>Great Antivirus!!!</td>\n",
       "      <td>This program has been keeping my computer runn...</td>\n",
       "      <td>2014</td>\n",
       "      <td>2014-05-14</td>\n",
       "      <td>Digital_Software</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>US</td>\n",
       "      <td>43410578</td>\n",
       "      <td>R2KX41Y1J5PPA6</td>\n",
       "      <td>B0064PFB9U</td>\n",
       "      <td>232554866</td>\n",
       "      <td>Office Mac Home and Student 2011 - 1PC/1User [...</td>\n",
       "      <td>3</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>N</td>\n",
       "      <td>Y</td>\n",
       "      <td>Does the job</td>\n",
       "      <td>It does the job but it's missing or rather, th...</td>\n",
       "      <td>2014</td>\n",
       "      <td>2014-05-14</td>\n",
       "      <td>Digital_Software</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>US</td>\n",
       "      <td>10783062</td>\n",
       "      <td>R3UPVHGBYFE6UQ</td>\n",
       "      <td>B004KPKSRQ</td>\n",
       "      <td>306022575</td>\n",
       "      <td>Pc Matic Performance Utility Suite (5 User Edi...</td>\n",
       "      <td>5</td>\n",
       "      <td>4</td>\n",
       "      <td>4</td>\n",
       "      <td>N</td>\n",
       "      <td>N</td>\n",
       "      <td>Love this new software that is keeping my syst...</td>\n",
       "      <td>I have the PC Matic on my comp, and couldn't b...</td>\n",
       "      <td>2014</td>\n",
       "      <td>2014-05-14</td>\n",
       "      <td>Digital_Software</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>US</td>\n",
       "      <td>51046135</td>\n",
       "      <td>R22P9WNTXY0VY6</td>\n",
       "      <td>B00FFIO0NA</td>\n",
       "      <td>386795082</td>\n",
       "      <td>TurboTax Premier Fed, Efile and State 2013</td>\n",
       "      <td>4</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>N</td>\n",
       "      <td>N</td>\n",
       "      <td>Same as Last Year</td>\n",
       "      <td>Performed as expected from years of experience...</td>\n",
       "      <td>2014</td>\n",
       "      <td>2014-05-14</td>\n",
       "      <td>Digital_Software</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  marketplace customer_id       review_id  product_id product_parent  \\\n",
       "0          US    18640556   RWKGA4N6R4KMY  B00C1QCBY4      640306152   \n",
       "1          US    11866626   RF8V3GF4MGT6M  B00H9A60O4      608720080   \n",
       "2          US    43410578  R2KX41Y1J5PPA6  B0064PFB9U      232554866   \n",
       "3          US    10783062  R3UPVHGBYFE6UQ  B004KPKSRQ      306022575   \n",
       "4          US    51046135  R22P9WNTXY0VY6  B00FFIO0NA      386795082   \n",
       "\n",
       "                                       product_title  star_rating  \\\n",
       "0                                   Sketchbook Pro 6            5   \n",
       "1               Avast Free Antivirus 2015 [Download]            5   \n",
       "2  Office Mac Home and Student 2011 - 1PC/1User [...            3   \n",
       "3  Pc Matic Performance Utility Suite (5 User Edi...            5   \n",
       "4         TurboTax Premier Fed, Efile and State 2013            4   \n",
       "\n",
       "   helpful_votes  total_votes vine verified_purchase  \\\n",
       "0              0            0    N                 Y   \n",
       "1              0            0    N                 N   \n",
       "2              0            0    N                 Y   \n",
       "3              4            4    N                 N   \n",
       "4              0            0    N                 N   \n",
       "\n",
       "                                     review_headline  \\\n",
       "0                                         Oustanding   \n",
       "1                                 Great Antivirus!!!   \n",
       "2                                       Does the job   \n",
       "3  Love this new software that is keeping my syst...   \n",
       "4                                  Same as Last Year   \n",
       "\n",
       "                                         review_body  year review_date  \\\n",
       "0  It's a much simpler application than Painter o...  2014  2014-05-14   \n",
       "1  This program has been keeping my computer runn...  2014  2014-05-14   \n",
       "2  It does the job but it's missing or rather, th...  2014  2014-05-14   \n",
       "3  I have the PC Matic on my comp, and couldn't b...  2014  2014-05-14   \n",
       "4  Performed as expected from years of experience...  2014  2014-05-14   \n",
       "\n",
       "   product_category  \n",
       "0  Digital_Software  \n",
       "1  Digital_Software  \n",
       "2  Digital_Software  \n",
       "3  Digital_Software  \n",
       "4  Digital_Software  "
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Execute statement using connection cursor\n",
    "cursor = connect(region_name=region_name, s3_staging_dir=s3_staging_dir).cursor()\n",
    "cursor.execute(statement)\n",
    "\n",
    "df = as_pandas(cursor)\n",
    "df.head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So, in just a few steps we have set up Amazon Athena to connect to our Amazon Customer Reviews TSV files, and transformed them into Apache Parquet file format. \n",
    "\n",
    "You might have noticed that our second sample query finished in a fraction of the time compared to the one before we ran on the TSV table. We speeded up our query results by leveraging our data being stored as Parquet and partitioned by `product_category`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_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.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}