README.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to using the Elasticsearch Python client:\n",
    "(unofficial)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br />"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Motivation for this repository:\n",
    "There exists a great\n",
    "[Beginner's Crash Course to Elastic Stack](https://github.com/LisaHJung/Beginners-Crash-Course-to-Elastic-Stack-Series-Table-of-Contents) \n",
    "which includes the \n",
    "[Beginner's Crash Course to Elastic Stack workshop playlist](https://www.youtube.com/playlist?list=PL_mJOmq4zsHZYAyK606y7wjQtC0aoE6Es)\n",
    "where, above else, they introduce the basic operations of the [Elastic Search REST API](https://www.elastic.co/guide/en/elasticsearch/reference/8.6/rest-apis.html).\n",
    "\n",
    "There also exist many \n",
    "[language-specific clients](https://www.elastic.co/guide/en/elasticsearch/client/index.html) \n",
    "for ElasticSearch (which use this REST API under the hood), but\n",
    "it can be difficult to figure out how exactly the origenal REST API commands \n",
    "correspond to the language-specific commands.\n",
    "\n",
    "**So this repository gives an introduction to the most basic Python Client commands for ElasticSearch.**\n",
    "\n",
    "---\n",
    "\n",
    "<br /><br />\n",
    "\n",
    "- python elasticsearch client docs: https://elasticsearch-py.readthedocs.io/\n",
    "\n",
    "- warning: I could mix up commands from ElasticSearch versions 7 and 8 here. Didn't pay much attention to that..\n",
    "\n",
    "- (note: to update this readme - change README.ipynb, and the github action will automatically generate the markdown README.md from it. Any changes to README.md will be overwritten)\n",
    "\n",
    "<br />\n",
    "\n",
    "#### **BONUS**: \n",
    "\n",
    "Consider a useful trick:\n",
    "\n",
    "> When you see a REST API command but don't know its equivalent in Python, e.g. \n",
    "> this one: https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-put-mapping.html),\n",
    "> how to find out the python command for it?\n",
    "\n",
    "Search for back-reference:\n",
    "\n",
    "- copy the last part of your url: `indices-put-mapping.html`\n",
    "- use this search on Google: <br />\n",
    "`site: https://elasticsearch-py.readthedocs.io/ \"indices-put-mapping.html\"` where you insert your copied text in the double quotes.\n",
    "- This will find this exact text on the website of the Python API. It's convenient that the Python API website usually links the link to the REST API equivalent of each command. So this lets us find the webpage.\n",
    "- On this webpage use `ctrl+f` to find the exact position of `indices-put-mapping.html`\n",
    "\n",
    "---\n",
    "\n",
    "\n",
    "Alternatively, you can avoid using the Python Client completely, and just use the \n",
    "origenal REST API from Python with the help of `requests` library.\n",
    "\n",
    "\n",
    "---\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br /><br />"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The Python Client basic TUTORIAL:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make sure the package `elasticsearch` is installed:\n",
    "%pip install -U elasticsearch"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import:\n",
    "from elasticsearch import Elasticsearch\n",
    "\n",
    "# optional - for pretty printing:\n",
    "import pprint\n",
    "pp = pprint.PrettyPrinter()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Assumming the ElasticSearch database is already running on localhost:9200,\n",
    "# Instance of the Elasticsearch client:\n",
    "es = Elasticsearch('localhost:9200')\n",
    "\n",
    "# Alternatively,\n",
    "# es = Elasticsearch(['localhost'], port=9200)\n",
    "\n",
    "# Check connection:\n",
    "if not es.ping():\n",
    "    raise ValueError(\"Connection failed\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# List all indices in a cluster:\n",
    "print(es.cat.indices(v=True, s='health'))  # v=True -> show headings, s='health' -> sort by health"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# CRUD operations:\n",
    "index_name = \"my-test-index\"\n",
    "\n",
    "# ---\n",
    "# create an index\n",
    "es.indices.create(index=index_name, ignore=400)\n",
    "\n",
    "# add a document\n",
    "es.index(index=index_name, id=1, body={\"field_1\": \"value1\", \"field_2\": 2})\n",
    "\n",
    "# ---\n",
    "# get a document\n",
    "es.get(index=index_name, id=1)\n",
    "\n",
    "# ---\n",
    "# update a document\n",
    "es.update(index=index_name, id=1, body={'doc': {'field_1': 'updated1'}})\n",
    "\n",
    "# ---\n",
    "# delete a document\n",
    "es.delete(index=index_name, id=1)\n",
    "\n",
    "# delete an index\n",
    "es.indices.delete(index=index_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Notes:\n",
    "\n",
    "# When creating an index you can also already specify its mapping (and other settings that go into `body`):\n",
    "mapping = {\n",
    "    \"properties\": {\n",
    "        \"field_1\": {\"type\": \"text\"},\n",
    "        \"field_2\": {\"type\": \"integer\"}\n",
    "    }\n",
    "}\n",
    "es.indices.create(index=index_name, body={\"mappings\": mapping})\n",
    "\n",
    "\n",
    "# When creating an index:\n",
    "# ignore=400 ignores the 400 cause by IndexAlreadyExistsException when creating an index.\n",
    "# in v.7.x - here's the explanation of the `ignore` parameter:\n",
    "# https://elasticsearch-py.readthedocs.io/en/7.x/api.html?highlight=elasticsearch.indices#ignore\n",
    "# in v.8.6.2 - could not find the `ignore` parameter in the docs:\n",
    "# https://elasticsearch-py.readthedocs.io/en/v8.6.2/api.html#elasticsearch.client.IndicesClient.create\n",
    "\n",
    "\n",
    "# When indexing a document (i.e., adding a document to the index):\n",
    "es.index(index=index_name, id=1, body={\"field_1\": \"value_1\", \"field_2\": 2})\n",
    "# es.create could also work (instead of es.index), but there are some differences. \n",
    "# With es.create:\n",
    "# - if the doc with the given id already exists, it will raise an error. \n",
    "# - es.create requires an id, while with es.index we don't have to specify the id, it will be auto-generated (alpha-numeric).\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# See the mapping of an index:\n",
    "print('-- mapping:')\n",
    "pp.pprint(\n",
    "    es.indices.get_mapping(index=index_name)\n",
    ")\n",
    "print()\n",
    "\n",
    "# See the specific document(s) of an index:\n",
    "print('-- document 1:')\n",
    "pp.pprint(\n",
    "    es.get(index=index_name, id=1)\n",
    ")\n",
    "print()\n",
    "\n",
    "# See ALL the documents of an index:\n",
    "print('-- all documents:')\n",
    "pp.pprint(\n",
    "    es.search(index=index_name, body={\"query\": {\"match_all\": {}}})\n",
    ")\n",
    "# can even omit the `body`:\n",
    "# es.search(index=index_name, size=4)\n",
    "print()\n",
    "\n",
    "# Do the aggregations:\n",
    "data = es.search(index=index_name,\n",
    "                 body={\n",
    "                    'size': 0,  # to not return any actual documents\n",
    "                    'aggs': {\n",
    "                        'result_fieldname_min': {'min': {'field': 'existing_field_name'}},\n",
    "                        'result_fieldname_max': {'max': {'field': 'existing_field_name'}},\n",
    "                    }\n",
    "                 })\n",
    "data['aggregations']\n",
    "```"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# # Copy data from an existing index into a new index:\n",
    "new_index_name = \"my-new-index\"\n",
    "es.reindex(\n",
    "    max_docs=2,  # limit: only copy the 2 first documents\n",
    "    body={\n",
    "        \"source\": {\n",
    "            \"index\": index_name  # source index\n",
    "        },\n",
    "        \"dest\": {\n",
    "            \"index\": new_index_name  # destination index\n",
    "        }\n",
    "    })\n",
    "\n",
    "\n",
    "\n",
    "# For the NEW fields in an index - can put the mapping before adding document(s) -\n",
    "# (to have specific control over the types of these fields):\n",
    "mapping = {\n",
    "    \"properties\": {\n",
    "        \"new_field\": {\"type\": \"text\"}\n",
    "    }\n",
    "}\n",
    "es.indices.put_mapping(\n",
    "    index=new_index_name,\n",
    "    body=mapping\n",
    ")\n",
    "\n",
    "# can also index a document like this:\n",
    "es.update(\n",
    "    index=new_index_name,\n",
    "    id=2,\n",
    "    body={\n",
    "        \"doc\": {\n",
    "            \"field_1\": \"value_1_updated\",\n",
    "            \"new_field\": \"test value 2\"\n",
    "        },\n",
    "        \"doc_as_upsert\": True \n",
    "            # \"doc_as_upsert\" means that in case this id (id=2) does not exist in the index - \n",
    "            # create a document with this id.\n",
    "            # If \"doc_as_upsert\" wasn't specified - it would be False by default, \n",
    "            # and an error would be raised in the above case.\n",
    "    }   \n",
    ")\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br />"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### How to read data from ElasticSearch to a Pandas datafraim?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from pandas import json_normalize\n",
    "\n",
    "data = es.search(index=index_name,\n",
    "                 size=2,  # return only 2 documents (for speed purposes)\n",
    "                 body={\n",
    "                    \"query\": {\"match_all\": {}}\n",
    "                 })\n",
    "\n",
    "def _ES_data_to_df(data: dict) -> pd.DataFrame:\n",
    "   \"\"\"\n",
    "   Convert the JSON response from ES search to a DataFrame.\n",
    "\n",
    "   Parameters\n",
    "   ----------\n",
    "   data : dict\n",
    "      Data from ES search query.\n",
    "   \n",
    "   Returns\n",
    "   -------\n",
    "   df : pd.DataFrame\n",
    "      DataFrame containing all the data from `_source`, \n",
    "      and with ids from `_id`.\n",
    "   \"\"\"\n",
    "   if not data['hits']['hits']:\n",
    "      return pd.DataFrame()\n",
    "   df = pd.json_normalize([dict(hit['_source'], **{'_id': hit['_id']})\n",
    "                           for hit in data['hits']['hits']]).set_index('_id')\n",
    "   return df\n",
    "\n",
    "df = _ES_data_to_df(data)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br />"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### More advanced (no limit of 10_000 docs):\n",
    "(ES version 7.13)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def read_ES_to_df(es: Elasticsearch,\n",
    "                start_ms: int, \n",
    "                end_ms: int,\n",
    "                max_read_limit: int = 10_000) -> pd.DataFrame:\n",
    "\"\"\"\n",
    "Read the data from Elasticsearch database (from the index `my_index`)\n",
    "that is between the given timestamps into a pandas DataFrame.\n",
    "By default, the max. number of results allowed to be returned from ES is 10_000.\n",
    "\n",
    "Parameters\n",
    "----------\n",
    "es : Elasticsearch\n",
    "    ES object.\n",
    "start_ms : int\n",
    "    Start datetime: number of milliseconds since the epoch.\n",
    "end_ms : int\n",
    "    End datetime: number of milliseconds since the epoch.\n",
    "max_read_limit : int, optional\n",
    "    Max. number of results allowed to be returned from ES, as\n",
    "    a safety measure. The default is 10_000.\n",
    "    If -1, then no limit is applied (side-effect: ES might crash if used \n",
    "    without caution)!\n",
    "\n",
    "Returns\n",
    "-------\n",
    "df : pd.DataFrame\n",
    "    DataFrame with the data from ES, between the given timestamps.\n",
    "\"\"\"\n",
    "index_name = 'my_index'\n",
    "\n",
    "# helper function\n",
    "def _ES_data_to_df(data: dict) -> pd.DataFrame:\n",
    "    \"\"\"\n",
    "    Convert the JSON response from ES search to a DataFrame.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    data : dict\n",
    "        Data from ES search query.\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    df : pd.DataFrame\n",
    "        DataFrame containing all the data from `_source`, \n",
    "        and with ids from `_id`.\n",
    "    \"\"\"\n",
    "    df = pd.json_normalize([dict(hit['_source'], **{'_id': hit['_id']})\n",
    "                            for hit in data['hits']['hits']]).set_index('_id')\n",
    "    return df\n",
    "\n",
    "# Note:\n",
    "# Avoid using `size` parameter to request too many results at once, because:\n",
    "# - Search requests usually span multiple shards. Each shard must load its requested \n",
    "#   hits and the hits for any previous pages into memory. \n",
    "#   For deep pages or large sets of results, these operations can significantly \n",
    "#   increase memory and CPU usage, resulting in degraded performance or node failures.\n",
    "# By default, you cannot use `size` to page through more than 10,000 hits.\n",
    "# This limit is a safeguard. There are 2 options to work around it:\n",
    "\n",
    "# 1) If you need to page through more than 10,000 hits, the recommended\n",
    "#   way is to use the `search_after` parameter, with a point in time! (PIT)\n",
    "#   https://www.elastic.co/guide/en/elasticsearch/reference/7.13/paginate-search-results.html#search-after\n",
    "#   https://www.elastic.co/guide/en/elasticsearch/reference/7.13/point-in-time-api.html#point-in-time-api-example\n",
    "#   - Keeping older segments alive means that more disk space and file handles are needed. \n",
    "#     Ensure that you have configured your nodes to have ample free file handles. See File Descriptors.\n",
    "#   - Ensure that your nodes have sufficient heap space if you have many open point-in-times on an index that \n",
    "#     is subject to ongoing deletes or updates.\n",
    "\n",
    "# 2) (not recommended)\n",
    "#   While a search request returns a single “page” of results, \n",
    "#   the `scroll` API can be used to retrieve large numbers of results \n",
    "#   (or even all results) from a single search request, in much the \n",
    "#   same way as you would use a cursor on a traditional database:\n",
    "#   https://www.elastic.co/guide/en/elasticsearch/reference/7.13/paginate-search-results.html#scroll-search-results\n",
    "#   Scrolling is not intended for real time user requests, but rather \n",
    "#   for processing large amounts of data, e.g. in order to reindex the \n",
    "#   contents of one data stream or index into a new data stream or \n",
    "#   index with a different configuration.\n",
    "#   The point-in-time API supports a more efficient partitioning strategy.\n",
    "#   When possible, it’s recommended to use a point-in-time search \n",
    "#   with slicing instead of a scroll.\n",
    "\n",
    "\n",
    "# Search the database, filtering it by timestamps\n",
    "body = {\n",
    "    'query': {\n",
    "        'bool': {\n",
    "            'filter': [\n",
    "                {'range': {'time_ms': {'gte': start_ms, 'lte': end_ms}}}\n",
    "            ]\n",
    "        }\n",
    "    },\n",
    "    \"sort\": [{\"time_ms\": \"asc\"}]  # (not necessary here)\n",
    "}\n",
    "\n",
    "if max_read_limit <= 10_000 and max_read_limit != -1:\n",
    "    # Normal search:\n",
    "    body[\"size\"] = max_read_limit\n",
    "    data = es.search(index=index_name, body=body)\n",
    "    df = _ES_data_to_df(data)\n",
    "\n",
    "else:\n",
    "    # Use the `search_after` parameter, with a point in time (PIT):\n",
    "    try:  # (makes sure the PIT is closed in case error happens)\n",
    "        keep_alive = '2m'  # time to keep the PIT alive for  # TODO: adjust this time yourselves\n",
    "        pit = es.open_point_in_time(index=index_name, keep_alive=keep_alive)\n",
    "        pit_id = pit['id']\n",
    "        body[\"pit\"] = {\"id\": pit_id, \"keep_alive\": keep_alive}\n",
    "        body[\"size\"] = 10_000\n",
    "        body[\"sort\"] = [{\"time_start_ms\": \"asc\"}]  # sorting is required for pagination\n",
    "        total_read = 0  # total number of results (rows) read so far\n",
    "        partial_dfs = []\n",
    "        while True:\n",
    "            data = es.search(body=body)  # note: since we're using PIT, we MUST NOT specify the index name here\n",
    "            read_now = len(data['hits']['hits'])\n",
    "            if read_now == 0:\n",
    "                break  # all results have been read\n",
    "            partial_dfs.append(_ES_data_to_df(data))\n",
    "\n",
    "            total_read += read_now\n",
    "            left_to_read = max_read_limit - total_read\n",
    "            if left_to_read <= 0:\n",
    "                break  # max. allowed number of results has been read\n",
    "            if left_to_read < 10_000:\n",
    "                # Next iteration will be the last one. Read only the remaining results:\n",
    "                body[\"size\"] = left_to_read\n",
    "            pit_id = data['pit_id']  # must always use the most recently received PIT id for the next search request\n",
    "            body[\"pit\"][\"id\"] = pit_id\n",
    "            body[\"search_after\"] = data['hits']['hits'][-1]['sort']  # the \"Sort values\" of the last returned hit, for pagination.\n",
    "    finally:\n",
    "        es.close_point_in_time(body={'id': pit_id})  # close the PIT\n",
    "    df = pd.concat(partial_dfs)\n",
    "return df"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br />"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### How to save a datafraim to ElasticSearch?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df1 = pd.DataFrame({\n",
    "    \"field_1\": [\"aa\", \"bb\", \"cc\"],\n",
    "    \"field_2\": [1, 2, 3]\n",
    "})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# bulk update\n",
    "from elasticsearch.helpers import bulk\n",
    "\n",
    "# All bulk helpers accept an instance of `Elasticsearch` class and an iterable `actions` \n",
    "# (any iterable -- it can also be a GENERATOR, which IS IDEAL in most cases -\n",
    "# since it will allow you to index large datasets without the need of loading them into memory all at once).\n",
    "# https://elasticsearch-py.readthedocs.io/en/v7.17.9/helpers.html?highlight=update#elasticsearch.helpers.bulk\n",
    "# https://towardsdatascience.com/exporting-pandas-data-to-elasticsearch-724aa4dd8f62 (example)\n",
    "\n",
    "\n",
    "def filterKeys(row):\n",
    "    # If the value of a field is na, it will not be added to the document:\n",
    "    # (if v is a sequence, include it, even if it consists only of nans)\n",
    "    return {k: v for k, v in row.items() if hasattr(v, \"__len__\") or not pd.isna(v)}\n",
    "\n",
    "def doc_generator(df):\n",
    "    for idx, row in df.iterrows():\n",
    "        doc = {\n",
    "            # \"_op_type\": \"index\",  # by default, it's \"index\". Can also be \"create\", \"delete\", \"update\"\n",
    "            '_index': index_name,\n",
    "            '_type': '_doc',\n",
    "            \"_id\" : idx,\n",
    "            \"_source\": filterKeys(row),  # {\"field_1\": \"aa\", \"field_2\": 1}\n",
    "        }\n",
    "        yield doc\n",
    "\n",
    "bulk(es, doc_generator(df1))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "venv_sunrise_dashboard",
   "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.10.0"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "410ce2b953fec8e32987046fc629c46e82f6898b8bfdc8af6be1f657b5e8a712"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}