examples/v3io-streams.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Using the Platform's Streaming API"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The platform's Streaming API enables working with data in the platform as streams.\n",
    "    For more information, see the platform's [streams](https://www.iguazio.com/docs/latest-release/data-layer/stream/) 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 python code on local machine that connects to a remote Iguazio platform, 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",
    "> You can get the access key from the platform dashboard: select the user-profile picture or icon from the top right corner of any page, and select **Access Keys** from the menu. In the **Access Keys** window, either copy an existing access key or create a new key and copy it. Alternatively, you can get the access key by checking the value of the `V3IO_ACCESS_KEY` environment variable in a web-shell or Jupyter Notebook service.\n",
    ">\n",
    "> For more information see the platform's [Data-Service Web-API General Structure](https://www.iguazio.com/docs/latest-release/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 [Put Multiple Records](#Put-Multiple-Records) 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/latest-release/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 stream:\n",
    "\n",
    "> **Note**: The following code uses the `V3IO_USERNAME` environment variable to store the data in the current user folder. When running python code on local machine that connects to a remote Iguazio platform, you should set this value to the user name you use for logging in to the system. Alternatively, you can get the user name by checking the value of the `V3IO_USERNAME` environment variable in a web-shell or Jupyter Notebook service."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from os import getenv, path\n",
    "\n",
    "V3IO_USERNAME = getenv('V3IO_USERNAME')\n",
    "STREAM_PATH = path.join(V3IO_USERNAME, 'data', 'v3io', 'stream')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a Stream"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Creates and configures a new stream. The new stream is available immediately upon its creation.\n",
    "\n",
    "You must configure the stream's shard count (the number of shards in the stream). You can increase the shard count at any time, but you cannot reduce it.\n",
    "\n",
    "You can also set the stream's retention period (default is 24 hours). After this period elapses, when new records are added to the stream, the earliest ingested records are deleted."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "204\n"
     ]
    }
   ],
   "source": [
    "response = v3io_client.stream.create(container=CONTAINER,\n",
    "                                     stream_path=STREAM_PATH,\n",
    "                                     shard_count=8)\n",
    "print(response.status_code)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Put Records"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `put` method to add records to a stream."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following example defines a function that converts text to lowercase words:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "146"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import re\n",
    "regex_non_lower_case = re.compile('[^a-z]')\n",
    "\n",
    "def text_to_words(text):\n",
    "    # split into words\n",
    "    words = text.split()\n",
    "    # convert to lower case\n",
    "    words = [w.lower() for w in words]\n",
    "    # Keep only letters\n",
    "    words = [regex_non_lower_case.sub('', w) for w in words]\n",
    "    \n",
    "    return words\n",
    "\n",
    "text1 = \"WOLF, meeting with a Lamb astray from the fold, resolved not to lay violent hands on him, but to find some plea to justify to the Lamb the Wolf’s right to eat him. He thus addressed him: “Sirrah, last year you grossly insulted me.” “Indeed,” bleated the Lamb in a mournful tone of voice, “I was not then born.” Then said the Wolf, “You feed in my pasture.” “No, good sir,” replied the Lamb, “I have not yet tasted grass.” Again said the Wolf, “You drink of my well.” “No,” exclaimed the Lamb, “I never yet drank water, for as yet my mother’s milk is both food and drink to me.” Upon which the Wolf seized him and ate him up, saying, “Well! I won’t remain supperless, even though you refute every one of my imputations.” The tyrant will always find a pretext for his tyranny.\"\n",
    "words1 = text_to_words(text1)\n",
    "len(words1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code converts the list of words to a record.\n",
    "A record is a list of dictionaries, where for each dictionary the `data` key contains the record data.\n",
    "The sample code displays the first 5 records:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[{'data': 'wolf'}, {'data': 'meeting'}, {'data': 'with'}, {'data': 'a'}, {'data': 'lamb'}]\n"
     ]
    }
   ],
   "source": [
    "records = [{'data': word} for word in words1]\n",
    "print(records[:5])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "200\n"
     ]
    }
   ],
   "source": [
    "response = v3io_client.stream.put_records(container=CONTAINER,\n",
    "                                          stream_path=STREAM_PATH,\n",
    "                                          records=records)\n",
    "print(response.status_code)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code writes another set of records to the stream:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "text2 = \"A BAT who fell upon the ground and was caught by a Weasel pleaded to be spared his life. The Weasel refused, saying that he was by nature the enemy of all birds. The Bat assured him that he was not a bird, but a mouse, and thus was set free. Shortly afterwards the Bat again fell to the ground and was caught by another Weasel, whom he likewise entreated not to eat him. The Weasel said that he had a special hostility to mice. The Bat assured him that he was not a mouse, but a bat, and thus a second time escaped. It is wise to turn circumstances to good account.\"\n",
    "response = v3io_client.stream.put_records(container=CONTAINER,\n",
    "                                          stream_path=STREAM_PATH,\n",
    "                                          records=[{'data': word} for word in text_to_words(text2)])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Get Records"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Multiple consumer instances can consume data from the same stream.\n",
    "A consumer retrieves records from a specific shard.\n",
    "It's recommended that you distribute the data consumption among several consumer instances (\"workers\"), assigning each instance one or more shards.\n",
    "\n",
    "For each shard, the consumer should determine the location within the shard from which to begin consuming records.\n",
    "This can be the earliest ingested record, the end of the shard, the first ingested record starting at a specific time, or a specific record identified by its sequence number (a unique record identifier that is assigned by the platform based on the record’s ingestion time).\n",
    "The consumer first uses a seek operation to obtain the desired consumption location, and then provides this location as the starting point for its record consumption.\n",
    "The consumption operation returns the location of the next record to consume within the shard, and this location should be used as the location for a subsequent consumption operation, allowing for sequential record consumption."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "shard_id = 0\n",
    "seek_response = v3io_client.stream.seek(container=CONTAINER,\n",
    "                                       stream_path=STREAM_PATH,\n",
    "                                       shard_id=shard_id,\n",
    "                                       seek_type='EARLIEST')\n",
    "location = seek_response.output.location"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `get_records` method to read from a specific shard of the stream (retrieve records).\n",
    "By default `get_records` limits the number of records returned to 1,000.\n",
    "For the sake of this demonstration, the sample code limits the number of returned records to 10 by setting the `limit` parameter.\n",
    "\n",
    "> **Note**: You will notice that some words are missing, this is because we sent different words on different shards and we query only shard '0' in this example."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['wolf', 'fold', 'him', 'to', 'him', 'you', 'in', 'not', 'feed', 'the']\n"
     ]
    }
   ],
   "source": [
    "get_response = v3io_client.stream.get_records(container=CONTAINER,\n",
    "                                              stream_path=STREAM_PATH,\n",
    "                                              shard_id=0,\n",
    "                                              location=location,\n",
    "                                              limit=10)\n",
    "words_shard0 = [record.data.decode('utf-8') for record in get_response.output.records]\n",
    "print(words_shard0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['again', 'well', 'drank', 'is', 'which', 'up', 'though', 'the', 'his', 'ground']\n"
     ]
    }
   ],
   "source": [
    "location = get_response.output.next_location\n",
    "\n",
    "get_response = v3io_client.stream.get_records(container=CONTAINER,\n",
    "                                          stream_path=STREAM_PATH,\n",
    "                                          shard_id=0,\n",
    "                                          location=location,\n",
    "                                          limit=10)\n",
    "words_shard0 = [record.data.decode('utf-8') for record in get_response.output.records]\n",
    "print(words_shard0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Describe Stream"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `describe` method to retrieve a stream's configuration, including the shard count and retention period."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "200\n"
     ]
    }
   ],
   "source": [
    "response = v3io_client.stream.describe(container=CONTAINER,\n",
    "                                       stream_path=STREAM_PATH)\n",
    "print(response.status_code)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Shards: 8\n",
      "Retention period: 24 hours\n"
     ]
    }
   ],
   "source": [
    "print(f'Shards: {response.output.shard_count}\\nRetention period: {response.output.retention_period_hours} hours')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Update Stream"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `update` method to updates a stream's configuration by increasing its shard count.\n",
    "The changes are applied immediately.\n",
    "\n",
    "> **Note**: You can increase the shard count at any time, but you cannot reduce it.\n",
    "> From the platform's perspective, there's virtually no cost to creating even thousands of shards.\n",
    "> However, if you increase a stream's shard count after its creation, new records with a previously used partition key will be assigned either to the same shard that was previously used for this partition key or to a new shard.\n",
    "> For more information see the platform's [stream sharding and partitioning](https://www.iguazio.com/docs/latest-release/data-layer/stream/#stream-sharding-and-partitioning) documentation.\n",
    "\n",
    "> **Note**: To use the streaming API to consume records from new shards after a shard-count increase, you must first restart the consumer application."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "204\n"
     ]
    }
   ],
   "source": [
    "response = v3io_client.stream.update(container=CONTAINER,\n",
    "                                     stream_path=STREAM_PATH,\n",
    "                                     shard_count=10)\n",
    "print(response.status_code)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Describe the stream again to see the updated shard count:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10\n"
     ]
    }
   ],
   "source": [
    "response = v3io_client.stream.describe(container=CONTAINER,\n",
    "                                       stream_path=STREAM_PATH)\n",
    "print(response.output.shard_count)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Assigning Shard IDs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the previous section, when you retrieved the stream records you didn't get all words.\n",
    "The reason is that by default, the platform assigns records to shards using a Round Robin algorithm, and the sample consumer code reads from a single shard.\n",
    "If you would like to ensure that a consumer gets all the words, you can optionally assign a record to specific stream shard by specifying a related shard ID by setting the `shard_id` value, or associate the record with a specific partition key to ensure that similar records are assigned to the same shard (by setting the `partition_key` value).\n",
    "For more information see the platform's [stream sharding and partitioning](https://www.iguazio.com/docs/latest-release/data-layer/stream/#stream-sharding-and-partitioning) documentation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's ignore the existing shards (shards 0 through 9) and increase the number of shards"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = v3io_client.stream.update(container=CONTAINER,\n",
    "                                     stream_path=STREAM_PATH,\n",
    "                                     shard_count=12)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Write the first text to shard 10:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = v3io_client.stream.put_records(container=CONTAINER,\n",
    "                                          stream_path=STREAM_PATH,\n",
    "                                          records=[{'data': word, 'shard_id': 10} for word in words1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And write the second text to shard 11:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = v3io_client.stream.put_records(container=CONTAINER,\n",
    "                                          stream_path=STREAM_PATH,\n",
    "                                          records=[{'data': word, 'shard_id': 11} for word in text_to_words(text2)])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, read from shard 10:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['wolf', 'meeting', 'with', 'a', 'lamb', 'astray', 'from', 'the', 'fold', 'resolved']...\n",
      "All words appear as expected\n"
     ]
    }
   ],
   "source": [
    "shard_id = 10\n",
    "seek_response = v3io_client.stream.seek(container=CONTAINER,\n",
    "                                   stream_path=STREAM_PATH,\n",
    "                                   shard_id=shard_id,\n",
    "                                   seek_type='EARLIEST')\n",
    "location = seek_response.output.location\n",
    "\n",
    "get_response = v3io_client.stream.get_records(container=CONTAINER,\n",
    "                                          stream_path=STREAM_PATH,\n",
    "                                          shard_id=shard_id,\n",
    "                                          location=location)\n",
    "words_shard10 = [record.data.decode('utf-8') for record in get_response.output.records]\n",
    "print(f'{words_shard10[:10]}...')\n",
    "if(words_shard10 == words1[:len(words_shard10)]):\n",
    "    print(\"All words appear as expected\")\n",
    "else:\n",
    "    print(\"Error, something is wrong\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Put Multiple Records"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "One way to increase performance is to 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. For instance, to set 16 parallel connections you should have in the beginning of the notebook `v3io_client = v3io.dataplane.Client(max_connections=16)`. The default is 8 connections."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> **Note**: The SDK also supports asynchronous API, which may also be useful to put multple objects. This capability is not demonstrated here, but you can read about it in the [v3io-py readme](https://github.com/v3io/v3io-py/blob/development/README.md#support-for-asyncio-experimental)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`put_records` accepts up to a maximum of 1,000 records.\n",
    "If the records limit is exceeded, the request fails.\n",
    "Therefore, you need to call `put_records` multiple times.\n",
    "The following example sends a large number of events.\n",
    "It creates a generator that returns a list of events.\n",
    "Each event has the following record:\n",
    "``` python\n",
    "'user': <user_id>\n",
    "'time': <event_time>\n",
    "'url': <url>\n",
    "```\n",
    "Each user is selected at random.\n",
    "The URL is also be a random string, and the `event_time` of the generated events is monotonically increased."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "import string\n",
    "from datetime import timedelta\n",
    "\n",
    "def generate_events(users, start_time, num_events, batch_size):\n",
    "    curr_time = start_time\n",
    "    \n",
    "    # letters to use for the URL generation \n",
    "    url_letters = string.ascii_lowercase + string.digits\n",
    "    \n",
    "    # Iterate through the number of events in batch_size increments\n",
    "    for i in range(0, num_events, batch_size):\n",
    "        events = []\n",
    "        for j in range(batch_size):\n",
    "            # choose a length to generate for the URL suffix\n",
    "            length = random.choice(range(20,30))\n",
    "            \n",
    "            # choose a user and the URL\n",
    "            event = {\n",
    "                'user': random.choice(users),\n",
    "                'time': curr_time,\n",
    "                'url': f\"https://example.com/{''.join(random.choice(url_letters) for i in range(length))}\"\n",
    "            }\n",
    "            \n",
    "            # increase the time for the next event randomly by a range of 0 to 10000 microseconds\n",
    "            curr_time += timedelta(microseconds=random.choice(range(0,10000)))\n",
    "            events.append(event)\n",
    "            \n",
    "        # yield returns the events as a generator, we will receive this in our loop\n",
    "        yield events"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Test the generator by printing a few events:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[{'user': 'user_3', 'time': datetime.datetime(2020, 10, 7, 5, 53, 0, 69161), 'url': 'https://example.com/rpoig8f1cbfno6b9m80o2'}, {'user': 'user_0', 'time': datetime.datetime(2020, 10, 7, 5, 53, 0, 78815), 'url': 'https://example.com/k1vrjnvgfygwwqc38hyf9sxm'}]\n",
      "[{'user': 'user_84', 'time': datetime.datetime(2020, 10, 7, 5, 53, 0, 79954), 'url': 'https://example.com/osfogyr3xkxwnrek8pk3'}, {'user': 'user_81', 'time': datetime.datetime(2020, 10, 7, 5, 53, 0, 86170), 'url': 'https://example.com/9oudocuzrenun5z3jqip98q1'}]\n"
     ]
    }
   ],
   "source": [
    "random.seed(42)\n",
    "from datetime import datetime\n",
    "\n",
    "for events in generate_events(users=[f'user_{index}' for index in range(0,100)],\n",
    "                              start_time=datetime(2020, 10, 7, 5, 53, 0, 69161),\n",
    "                              num_events=4,\n",
    "                              batch_size=2):\n",
    "    print(events)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Seek the latest position in one of the shards, in order to later retrieve the new data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "shard_id = 0\n",
    "seek_response = v3io_client.stream.seek(container=CONTAINER,\n",
    "                                   stream_path=STREAM_PATH,\n",
    "                                   shard_id=shard_id,\n",
    "                                   seek_type='LATEST')\n",
    "location = seek_response.output.location"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "num_users = 10000\n",
    "num_events = 100000\n",
    "batch_size = 1000\n",
    "\n",
    "for events in  generate_events(users=[f'user_{index}' for index in range(num_users)],\n",
    "                               start_time=datetime.now(),\n",
    "                               num_events=num_events,\n",
    "                               batch_size=batch_size):\n",
    "    records = [{'data': json.dumps(event, default=str), 'partition_key': event.get('user')} for event in events]\n",
    "    \n",
    "    v3io_client.batch.stream.put_records(container=CONTAINER,\n",
    "                                         stream_path=STREAM_PATH,\n",
    "                                         records=records)\n",
    "\n",
    "# wait for all writes to complete\n",
    "responses = v3io_client.batch.wait()    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The looped `put_records` interface in the previous code block sends all `put_records` 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": [
    "Test the received record from the previously obtained location:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Received 1000 records\n",
      "First record: {'user': 'user_7491', 'time': '2022-09-15 17:53:54.679908', 'url': 'https://example.com/uea3ge8n6qiwepxsk28t7a9t'}\n"
     ]
    }
   ],
   "source": [
    "get_response = v3io_client.stream.get_records(container=CONTAINER,\n",
    "                                          stream_path=STREAM_PATH,\n",
    "                                          shard_id=shard_id,\n",
    "                                          location=location)\n",
    "records = get_response.output.records\n",
    "\n",
    "print(f'Received {len(records)} records')\n",
    "print(f'First record: {json.loads(get_response.output.records[0].data)}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Delete Stream"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Deletes a stream object along with all of its shards."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "204\n"
     ]
    }
   ],
   "source": [
    "response = v3io_client.stream.delete(container=CONTAINER, stream_path=STREAM_PATH)\n",
    "print(response.status_code)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alternatively, when running on the Iguazio platform (not from remote), you can use the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import shutil\n",
    "V3IO_STREAM_PATH = path.join(path.sep, 'v3io', CONTAINER, STREAM_PATH)\n",
    "shutil.rmtree(V3IO_STREAM_PATH)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Or use this command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!rm -r $V3IO_STREAM_PATH"
   ]
  }
 ],
 "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"
  },
  "vscode": {
   "interpreter": {
    "hash": "916dbcbb3f70747c44a77c7bcd40155683ae19c65e1c03b4aa3499c5328201f1"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}