ingest.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "31508664",
   "metadata": {},
   "source": [
    "More info at https://github.com/Azure/azure-kusto-python/blob/master/azure-kusto-ingest/tests/sample.py\n",
    "\n",
    "Before running make sure that:\n",
    "* You have ingestor access to the ADX Database\n",
    "* A table exist matching your data\n",
    "\n",
    "TODO\n",
    "\n",
    "- Add code for table creation\n",
    "- nbdev"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "54eefeef",
   "metadata": {},
   "source": [
    "### Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ee34e313",
   "metadata": {},
   "outputs": [],
   "source": [
    "import io\n",
    "import os\n",
    "import requests\n",
    "\n",
    "from azure.kusto.data import KustoConnectionStringBuilder\n",
    "from azure.kusto.ingest import (\n",
    "    QueuedIngestClient,\n",
    "    IngestionProperties,\n",
    "    FileDescriptor,\n",
    "    BlobDescriptor,\n",
    "    StreamDescriptor,\n",
    "    DataFormat,\n",
    "    ReportLevel,\n",
    "    IngestionMappingType,\n",
    "    KustoStreamingIngestClient,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7b5ed7d2",
   "metadata": {},
   "source": [
    "### Config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "14632a24",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Note the ingest- prefix\n",
    "cluster = \"https://ingest-dataexplorerda.northeurope.kusto.windows.net\"\n",
    "\n",
    "# The file to ingest\n",
    "filename = \"path/to/csv/0\" # without file ending\n",
    "\n",
    "# DB and table name\n",
    "database_name = \"db-drilling-analytics\"\n",
    "table_name = \"test_table\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c57c2af0",
   "metadata": {},
   "source": [
    "### ConnectionString"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9eeca0c4",
   "metadata": {},
   "outputs": [],
   "source": [
    "#kcsb = KustoConnectionStringBuilder.with_aad_device_authentication(cluster, aad_tenant_id)\n",
    "#kcsb.authority_id = aad_tenant_id\n",
    "\n",
    "# In case you want to authenticate with a System Assigned Managed Service Identity (MSI)\n",
    "#kcsb = KustoConnectionStringBuilder.with_aad_managed_service_identity_authentication(cluster)\n",
    "\n",
    "# In case you want to authenticate with AAD device code.\n",
    "# Please note that if you choose this option, you'll need to autenticate for every new instance that is initialized.\n",
    "# It is highly recommended to create one instance and use it for all of your queries.\n",
    "kcsb = KustoConnectionStringBuilder.with_aad_device_authentication(cluster)\n",
    "\n",
    "# there are more options for authenticating - see azure-kusto-data samples"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e82f10dc",
   "metadata": {},
   "source": [
    "### Client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "643121a2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# The authentication method will be taken from the chosen KustoConnectionStringBuilder.\n",
    "client = QueuedIngestClient(kcsb)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "989d94ab",
   "metadata": {},
   "source": [
    "### Ingestion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9d2c4e6f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# there are a lot of useful properties, make sure to go over docs and check them out\n",
    "ingestion_props = IngestionProperties(\n",
    "    database=f\"{database_name}\",\n",
    "    table=f\"{table_name}\",\n",
    "    data_format=DataFormat.CSV,\n",
    "    # in case status update for success are also required\n",
    "    report_level=ReportLevel.FailuresAndSuccesses,\n",
    "    #ignore_first_record = True,\n",
    "    # in case a mapping is required\n",
    "    # ingestion_mapping_reference=\"{json_mapping_that_already_exists_on_table}\"\n",
    "    # ingestion_mapping_type=IngestionMappingType.JSON\n",
    ")\n",
    "\n",
    "# ingest from file\n",
    "file_descriptor = FileDescriptor(f\"{filename}.csv\", os.path.getsize(f\"{filename}.csv\"))  #  is the raw size of the data in bytes.\n",
    "client.ingest_from_file(file_descriptor, ingestion_properties=ingestion_props)\n",
    "\n",
    "# OR\n",
    "#client.ingest_from_file(f\"{filename}.csv\", ingestion_properties=ingestion_props)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "742ad082",
   "metadata": {},
   "source": [
    "## More options"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6b2407e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Optional 1: ingest from other formats\n",
    "\n",
    "\n",
    "# ingest from blob\n",
    "blob_descriptor = BlobDescriptor(\n",
    "    \"https://{path_to_blob}.csv.gz?sp=rl&st=2020-05-20T13:38:37Z&se=2020-05-21T13:38:37Z&sv=2019-10-10&sr=c&sig=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\",\n",
    "    10,\n",
    ")  # 10 is the raw size of the data in bytes.\n",
    "client.ingest_from_blob(blob_descriptor, ingestion_properties=ingestion_props)\n",
    "\n",
    "\n",
    "# ingest from dataframe\n",
    "import pandas\n",
    "\n",
    "fields = [\"id\", \"name\", \"value\"]\n",
    "rows = [[1, \"abc\", 15.3], [2, \"cde\", 99.9]]\n",
    "\n",
    "df = pandas.DataFrame(data=rows, columns=fields)\n",
    "\n",
    "client.ingest_from_dataframe(df, ingestion_properties=ingestion_props)\n",
    "\n",
    "# ingest a whole folder.\n",
    "import os\n",
    "\n",
    "path = \"folder/path\"\n",
    "[client.ingest_from_file(f, ingestion_properties=ingestion_props) for f in os.listdir(path)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7621f477",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Optional 2: view ingestion status\n",
    "\n",
    "##################################################################\n",
    "##                        INGESTION STATUS                      ##\n",
    "##################################################################\n",
    "\n",
    "# if status updates are required, something like this can be done\n",
    "import pprint\n",
    "import time\n",
    "from azure.kusto.ingest.status import KustoIngestStatusQueues\n",
    "\n",
    "qs = KustoIngestStatusQueues(client)\n",
    "\n",
    "MAX_BACKOFF = 180\n",
    "\n",
    "backoff = 1\n",
    "while True:\n",
    "    ################### NOTICE ####################\n",
    "    # in order to get success status updates,\n",
    "    # make sure ingestion properties set the\n",
    "    # reportLevel=ReportLevel.FailuresAndSuccesses.\n",
    "    if qs.success.is_empty() and qs.failure.is_empty():\n",
    "        time.sleep(backoff)\n",
    "        backoff = min(backoff * 2, MAX_BACKOFF)\n",
    "        print(\"No new messages. backing off for {} seconds\".format(backoff))\n",
    "        continue\n",
    "\n",
    "    backoff = 1\n",
    "\n",
    "    success_messages = qs.success.pop(10)\n",
    "    failure_messages = qs.failure.pop(10)\n",
    "\n",
    "    pprint.pprint(\"SUCCESS : {}\".format(success_messages))\n",
    "    pprint.pprint(\"FAILURE : {}\".format(failure_messages))\n",
    "\n",
    "    # you can of course separate them and dump them into a file for follow up investigations\n",
    "    with open(\"successes.log\", \"w+\") as sf:\n",
    "        for sm in success_messages:\n",
    "            sf.write(str(sm))\n",
    "\n",
    "    with open(\"failures.log\", \"w+\") as ff:\n",
    "        for fm in failure_messages:\n",
    "            ff.write(str(fm))\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a46bed6a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Optional 3: Streaming ingest\n",
    "\n",
    "# Authenticate against this cluster endpoint as shows in the Auth section\n",
    "cluster = \"https://{cluster_name}.kusto.windows.net\"\n",
    "\n",
    "client = KustoStreamingIngestClient(kcsb)\n",
    "\n",
    "ingestion_properties = IngestionProperties(database=\"{database_name}\", table=\"{table_name}\", data_format=DataFormat.CSV)\n",
    "\n",
    "# ingest from file\n",
    "file_descriptor = FileDescriptor(\"{filename}.csv\", 3333)  # 3333 is the raw size of the data in bytes.\n",
    "client.ingest_from_file(file_descriptor, ingestion_properties=ingestion_properties)\n",
    "client.ingest_from_file(\"{filename}.csv\", ingestion_properties=ingestion_properties)\n",
    "\n",
    "# ingest from dataframe\n",
    "import pandas\n",
    "\n",
    "fields = [\"id\", \"name\", \"value\"]\n",
    "rows = [[1, \"abc\", 15.3], [2, \"cde\", 99.9]]\n",
    "\n",
    "df = pandas.DataFrame(data=rows, columns=fields)\n",
    "\n",
    "client.ingest_from_dataframe(df, ingestion_properties=ingestion_properties)\n",
    "\n",
    "# ingest from stream\n",
    "byte_sequence = b\"56,56,56\"\n",
    "bytes_stream = io.BytesIO(byte_sequence)\n",
    "client.ingest_from_stream(bytes_stream, ingestion_properties=ingestion_properties)\n",
    "\n",
    "stream_descriptor = StreamDescriptor(bytes_stream)\n",
    "client.ingest_from_stream(stream_descriptor, ingestion_properties=ingestion_properties)\n",
    "\n",
    "str_sequence = u\"57,57,57\"\n",
    "str_stream = io.StringIO(str_sequence)\n",
    "client.ingest_from_stream(str_stream, ingestion_properties=ingestion_properties)"
   ]
  }
 ],
 "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}