data-ingestion-and-preparation/v3io-kv.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Using the Platform's NoSQL (Key-Value) API"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The platform's NoSQL (a.k.a. \"key-value\"/\"KV\") API provides access to the NoSQL database service, which enables storing and consuming data in a tabular format.\n",
    "For more information, see the platform's [NoSQL-databases](https://www.iguazio.com/docs/v3.0/data-layer/nosql/) documentation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initialize"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import v3io.dataplane"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a dataplane client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "v3io_client = v3io.dataplane.Client()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> **Note**: You can pass to the client the `endpoint` and `access_key` parameters explicitly.\n",
    "> The following code is equivalent to the default values:\n",
    ">\n",
    "> ``` python\n",
    "> from os import getenv\n",
    "> v3io_client = v3io.dataplane.Client(endpoint='http://v3io-webapi:8081',\n",
    ">                                     access_key=getenv('V3IO_ACCESS_KEY'))\n",
    "> ```\n",
    ">\n",
    "> When running the code remotely, you can obtain the URL of your cluster by copying the API URL of the web-APIs service (`webapi`) from the **Services** dashboard page. You can select between two types of URLs:\n",
    ">\n",
    "> - **HTTPS Direct** (recommended) &mdash; a URL of the format `https://<tenant IP>:<web-APIs port>`; for example, `https://default-tenant.app.mycluster.iguazio.com:8443`.\n",
    "> - **HTTPS** &mdash; a URL of the format `https://webapi.<tenant IP>`; for example, `https://webapi.default-tenant.app.mycluster.iguazio.com`.\n",
    ">\n",
    "> For more information see the platform's [Data-Service Web-API General Structure](https://www.iguazio.com/docs/v3.0/data-layer/reference/web-apis/data-service-web-api-gen-struct/) documentation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> **Number of maximum parallel connections**: Another noteworthy parameter is `max_connections`, which defines the number of maximum parallel connections when performing batch operations.\n",
    "> If left unspecified, the default is 8 connections.\n",
    "> For more information see the [Write Multiple Items](#Write-Multiple-Items) section in this tutorial."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Set the Data Path"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "All data in the platform is stored in user-defined data containers.\n",
    "This tutorial uses the predefined \"users\" container.\n",
    "For more information refer to the platform's [data-containers](https://www.iguazio.com/docs/v3.0/data-layer/containers/) documentation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "CONTAINER = 'users'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Set the data path for storing the NoSQL (KV) table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from os import getenv, path\n",
    "\n",
    "V3IO_USERNAME = getenv('V3IO_USERNAME')\n",
    "TABLE_PATH = path.join(V3IO_USERNAME, 'examples', 'v3io', 'kv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Write an Item"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `write` method to create an item with the provided attributes.\n",
    "If an item with the same name (primary key) already exists in the specified table, the existing item is completely overwritten (replaced with a new item).\n",
    "If the item or table don't exist, the operation creates them.\n",
    "\n",
    "> **Note**: NoSQL tables in the platform don't need to be created prior to ingestion.\n",
    "> When writing data to a NoSQL table, if the table doesn't exit, it's automatically created in the specified path as part of the write operation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create an example item:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "from array import array\n",
    "item = {\n",
    "    'title': \"The Godfather\",\n",
    "    'rating': 9.2,\n",
    "    'release_date': datetime(1972, 3, 24),\n",
    "    'duration': 175\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Write to the NoSQL (KV) storage:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing to iguazio/examples/v3io/kv\n",
      "Status code: 200\n"
     ]
    }
   ],
   "source": [
    "print(f'Writing to {TABLE_PATH}')\n",
    "response = v3io_client.kv.put(container=CONTAINER, table_path=TABLE_PATH, key='tt0068646', attributes=item)\n",
    "print(f'Status code: {response.status_code}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read an Item"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `get` method to retrieves the requested attributes of a table item."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = v3io_client.kv.get(container=CONTAINER, table_path=TABLE_PATH, key='tt0068646')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Print the response output item:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'title': 'The Godfather', 'rating': 9.2, 'release_date': datetime.datetime(1972, 3, 24, 0, 0), 'duration': 175}\n"
     ]
    }
   ],
   "source": [
    "print(response.output.item)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Update an Item"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `update` method to update the attributes of a table item.\n",
    "If the specified item or table don't exist, the operation creates them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "200\n"
     ]
    }
   ],
   "source": [
    "response = v3io_client.kv.update(container=CONTAINER, table_path=TABLE_PATH, key='tt0068646', attributes={'rating': 9.3})\n",
    "print(response.status_code)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Delete an Item"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "204\n"
     ]
    }
   ],
   "source": [
    "response = v3io_client.kv.delete(container=CONTAINER, table_path=TABLE_PATH, key='tt0068646')\n",
    "print(response.status_code)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Write Multiple Items"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To get the highest possible throughput, you can send many requests towards the data layer and wait for all the responses to arrive (rather than send each request and wait for the response).\n",
    "The SDK supports this through batching.\n",
    "Any API call can be made through the client's built in `batch` object.\n",
    "The API call receives the exact same arguments it would normally receive (except for `raise_for_status`), and does not block until the response arrives.\n",
    "To wait for all pending responses, call the `wait` method of the `batch` object."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> **Note**: The number of parallel connections is determined by the `max_connections` parameter when you created the client.\n",
    "> For instance, to set 16 parallel connections you should have in the beginning of the notebook `v3io_client = v3io.dataplane.Client(max_connections=16)`.\n",
    "> The default is 8 connections."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "movies = [\n",
    "{'key': \"tt0111161\",\n",
    " 'item': {'title': \"The Shawshank Redemption\",                          'rating': 9.2, 'release_date': datetime(1994, 10, 14), 'duration': 142}},\n",
    "{'key': \"tt0068646\",\n",
    " 'item': {'title': \"The Godfather\",                                     'rating': 9.1, 'release_date': datetime(1972, 3, 24),  'duration': 175}},\n",
    "{'key': \"tt0071562\",\n",
    " 'item': {'title': \"The Godfather: Part II\",                            'rating': 9,   'release_date': datetime(1974, 12, 18), 'duration': 202}},\n",
    "{'key': \"tt0468569\",\n",
    " 'item': {'title': \"The Dark Knight\",                                   'rating': 9,   'release_date': datetime(2008, 7, 18),  'duration': 152}},\n",
    "{'key': \"tt0050083\",\n",
    " 'item': {'title': \"12 Angry Men\",                                      'rating': 8.9, 'release_date': datetime(1957, 4, 10),  'duration': 96}},\n",
    "{'key': \"tt0108052\",\n",
    " 'item': {'title': \"Schindler's List\",                                  'rating': 8.9, 'release_date': datetime(1993, 2, 4),   'duration': 195}},\n",
    "{'key': \"tt0167260\",\n",
    " 'item': {'title': \"The Lord of the Rings: The Return of the King\",     'rating': 8.9, 'release_date': datetime(2003, 12, 17), 'duration': 201}},\n",
    "{'key': \"tt0110912\",\n",
    " 'item': {'title': \"Pulp Fiction\",                                      'rating': 8.8, 'release_date': datetime(1994, 10, 14), 'duration': 154}},\n",
    "{'key': \"tt0060196\",\n",
    " 'item': {'title': \"The Good, the Bad and the Ugly\",                    'rating': 8.8, 'release_date': datetime(1967, 12, 29), 'duration': 178}},\n",
    "{'key': \"tt0120737\",\n",
    " 'item': {'title': \"The Lord of the Rings: The Fellowship of the Ring\", 'rating': 8.8, 'release_date': datetime(2001, 12, 19), 'duration': 178}}\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "for movie in movies:\n",
    "    v3io_client.batch.kv.put(container=CONTAINER, table_path=TABLE_PATH, key=movie.get('key'), attributes=movie.get('item'))\n",
    "\n",
    "# wait for all writes to complete\n",
    "responses = v3io_client.batch.wait()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The looped `put` interface in the previous code block sends all `put` requests to the data layer in parallel.\n",
    "When `wait` is called, it blocks until either all responses arrive &mdash; in which case it returns a `Responses` object that contains the `responses` of each call &mdash; or an error occurs &mdash; in which case an exception is thrown.\n",
    "You can pass `raise_for_status` to `wait`, and it behaves as previously explained.\n",
    "\n",
    "> **Note:** The `batch` object is stateful, therefore you can only create one batch at a time.\n",
    "> However, you can create multiple parallel batches yourself through the client's `create_batch` interface."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read Multiple Items"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Retrieves (reads) attributes of multiple items in a table, according to the specified criteria."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'title': 'Pulp Fiction', 'rating': 8.8}\n",
      "{'title': '12 Angry Men', 'rating': 8.9}\n",
      "{'title': 'The Shawshank Redemption', 'rating': 9.2}\n",
      "{'title': 'The Dark Knight', 'rating': 9}\n"
     ]
    }
   ],
   "source": [
    "items_cursor = v3io_client.kv.new_cursor(container=CONTAINER,\n",
    "                                         table_path=TABLE_PATH,\n",
    "                                         attribute_names=['title', 'rating'],\n",
    "                                         filter_expression='duration < 170')\n",
    "\n",
    "for item in items_cursor.all():\n",
    "    print(item)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a Schema (Optional)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To support reading and writing NoSQL data using structured-data interfaces &mdash; such as Spark DataFrames, Presto, and [V3IO Frames](https://www.iguazio.com/docs/v3.0/data-layer/reference/frames/) (\"Frames\") &mdash; the platform uses a schema file that defines the schema of the data structure.\n",
    "When writing NoSQL data in the platform using a Spark or Frames DataFrame, the schema of the data table is automatically identified and saved and then retrieved when using a structure-data interface to read data from the same table (unless you explicitly define the schema for the read operation).\n",
    "However, to use a structure-data interface to read NoSQL data that was not written in this manner, you first need to define the table schema:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "fields = [\n",
    "    {\n",
    "        'name': 'title',\n",
    "        'type': 'string',\n",
    "        'nullable': False\n",
    "    },\n",
    "    {\n",
    "        'name': 'rating',\n",
    "        'type': 'double',\n",
    "        'nullable': True        \n",
    "    },\n",
    "    {\n",
    "        'name': 'release_date',\n",
    "        'type': 'timestamp',\n",
    "        'nullable': False\n",
    "    },\n",
    "    {\n",
    "        'name': 'duration',\n",
    "        'type': 'long',\n",
    "        'nullable': False        \n",
    "    }\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "204\n"
     ]
    }
   ],
   "source": [
    "repsonse = v3io_client.kv.create_schema(container=CONTAINER, table_path=TABLE_PATH, key='title', fields=fields)\n",
    "print(response.status_code)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Read the KV table using Frames:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "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>duration</th>\n",
       "      <th>rating</th>\n",
       "      <th>release_date</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>title</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>The Dark Knight</th>\n",
       "      <td>152</td>\n",
       "      <td>9.0</td>\n",
       "      <td>2008-07-18 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>The Godfather: Part II</th>\n",
       "      <td>202</td>\n",
       "      <td>9.0</td>\n",
       "      <td>1974-12-18 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>Schindler's List</th>\n",
       "      <td>195</td>\n",
       "      <td>8.9</td>\n",
       "      <td>1993-02-04 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>12 Angry Men</th>\n",
       "      <td>96</td>\n",
       "      <td>8.9</td>\n",
       "      <td>1957-04-10 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>The Lord of the Rings: The Fellowship of the Ring</th>\n",
       "      <td>178</td>\n",
       "      <td>8.8</td>\n",
       "      <td>2001-12-19 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>The Godfather</th>\n",
       "      <td>175</td>\n",
       "      <td>9.1</td>\n",
       "      <td>1972-03-24 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>The Shawshank Redemption</th>\n",
       "      <td>142</td>\n",
       "      <td>9.2</td>\n",
       "      <td>1994-10-14 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>The Lord of the Rings: The Return of the King</th>\n",
       "      <td>201</td>\n",
       "      <td>8.9</td>\n",
       "      <td>2003-12-17 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>The Good, the Bad and the Ugly</th>\n",
       "      <td>178</td>\n",
       "      <td>8.8</td>\n",
       "      <td>1967-12-29 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>Pulp Fiction</th>\n",
       "      <td>154</td>\n",
       "      <td>8.8</td>\n",
       "      <td>1994-10-14 00:00:00+00:00</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                   duration  rating  \\\n",
       "title                                                                 \n",
       "The Dark Knight                                         152     9.0   \n",
       "The Godfather: Part II                                  202     9.0   \n",
       "Schindler's List                                        195     8.9   \n",
       "12 Angry Men                                             96     8.9   \n",
       "The Lord of the Rings: The Fellowship of the Ring       178     8.8   \n",
       "The Godfather                                           175     9.1   \n",
       "The Shawshank Redemption                                142     9.2   \n",
       "The Lord of the Rings: The Return of the King           201     8.9   \n",
       "The Good, the Bad and the Ugly                          178     8.8   \n",
       "Pulp Fiction                                            154     8.8   \n",
       "\n",
       "                                                               release_date  \n",
       "title                                                                        \n",
       "The Dark Knight                                   2008-07-18 00:00:00+00:00  \n",
       "The Godfather: Part II                            1974-12-18 00:00:00+00:00  \n",
       "Schindler's List                                  1993-02-04 00:00:00+00:00  \n",
       "12 Angry Men                                      1957-04-10 00:00:00+00:00  \n",
       "The Lord of the Rings: The Fellowship of the Ring 2001-12-19 00:00:00+00:00  \n",
       "The Godfather                                     1972-03-24 00:00:00+00:00  \n",
       "The Shawshank Redemption                          1994-10-14 00:00:00+00:00  \n",
       "The Lord of the Rings: The Return of the King     2003-12-17 00:00:00+00:00  \n",
       "The Good, the Bad and the Ugly                    1967-12-29 00:00:00+00:00  \n",
       "Pulp Fiction                                      1994-10-14 00:00:00+00:00  "
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import v3io_frames as v3f\n",
    "\n",
    "v3f_client = v3f.Client('framesd:8081', container=CONTAINER)\n",
    "v3f_client.read(backend='kv', table=TABLE_PATH)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Delete the Table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Currently, most platform APIs don't have a dedicated method for deleting a table.\n",
    "An exception to this is the V3IO Frames Client class, which supports a delete method for the NoSQL backend; for more information, see the [Frames documentation](https://www.iguazio.com/docs/v3.0/data-layer/reference/frames/).\n",
    "However, you can use the file-system interface to delete a table directory from the relevant data container:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "from os import sep\n",
    "import shutil\n",
    "V3IO_TABLE_PATH = path.join(sep, 'v3io', CONTAINER, TABLE_PATH)\n",
    "shutil.rmtree(V3IO_TABLE_PATH)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alternatively you can use the following commands:\n",
    "```\n",
    "!rm -r $V3IO_TABLE_PATH\n",
    "```"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}