examples/BulkImport.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "c5ded82d-d20e-4afd-b8f3-d5fbd8cfd821",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Remotely Using the BulkImport API via PyArrow\n",
    "\n",
    "The BulkImport API (the underpinnings of `neo4j-admin import`) provides the fastest way to bootstrap a new database because it performs construction of the record store offline (without transactions).\n",
    "\n",
    "## Challenges with `neo4j-admin import`\n",
    "\n",
    "The tooling presupposes:\n",
    "\n",
    "1. Access to the filesystem on the Neo4j server 😬\n",
    "2. Data formatted into CSVs 🤮\n",
    "\n",
    "## The PyArrow / `neo4j-arrow` Solution\n",
    "\n",
    "1. Remotely stream data (nodes and edges) to the `neo4j-arrow` service from any client.\n",
    "2. Data can be in any Arrow-compatible format (Pandas DataFrames, Python dicts, Parquet, CSV, JSON, etc.)\n",
    "\n",
    "## Demo Time!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "ffed471e-511a-4a46-991a-0402ff485366",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[ActionType(type='import.bulk', description='Use neo4j bulk import to bootstrap a new database.')]"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import neo4j_arrow as na\n",
    "\n",
    "client = na.Neo4jArrow('neo4j', 'password', ('voutila-arrow-test', 9999), tls=True, verify_tls=False)\n",
    "[action for action in client.list_actions() if action[0] == 'import.bulk']"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a6bc3e0a-15f8-4adb-b751-aa9029f53d59",
   "metadata": {},
   "source": [
    "### Let's set up some data...\n",
    "`pyimport` is some scripting around reading a directory of CSVs created via `gds.beta.graph.export.csv`.\n",
    "\n",
    "It has some caveats, but fine for demo purposes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "83dc2ebe-d273-4734-a341-2cc0d2221450",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Loading from ./random\n",
      "import targets: {'nodes_', 'relationships_SIMILAR_', 'relationships_REL_'}\n",
      "files = ['nodes_0.csv', 'nodes_2.csv', 'nodes_3.csv', 'nodes_1.csv']\n",
      "fields = [Field(name='ID', type=<FieldType.NODE_ID: 'ID'>, id_space='Global'), Field(name='louvain', type=<FieldType.LONG: 'long'>, id_space='Global')]\n",
      "\n",
      "reading ./random/nodes_0.csv...\n",
      "reading ./random/nodes_2.csv...\n",
      "reading ./random/nodes_3.csv...\n",
      "reading ./random/nodes_1.csv...\n",
      "files = ['relationships_SIMILAR_2.csv', 'relationships_SIMILAR_0.csv', 'relationships_SIMILAR_3.csv', 'relationships_SIMILAR_1.csv']\n",
      "fields = [Field(name='START_ID', type=<FieldType.START_ID: 'START_ID'>, id_space='Global'), Field(name='END_ID', type=<FieldType.END_ID: 'END_ID'>, id_space='Global'), Field(name='score', type=<FieldType.DOUBLE: 'double'>, id_space='Global')]\n",
      "\n",
      "reading ./random/relationships_SIMILAR_2.csv...\n",
      "reading ./random/relationships_SIMILAR_0.csv...\n",
      "reading ./random/relationships_SIMILAR_3.csv...\n",
      "reading ./random/relationships_SIMILAR_1.csv...\n",
      "files = ['relationships_REL_1.csv', 'relationships_REL_2.csv', 'relationships_REL_0.csv', 'relationships_REL_3.csv']\n",
      "fields = [Field(name='START_ID', type=<FieldType.START_ID: 'START_ID'>, id_space='Global'), Field(name='END_ID', type=<FieldType.END_ID: 'END_ID'>, id_space='Global')]\n",
      "\n",
      "reading ./random/relationships_REL_1.csv...\n",
      "reading ./random/relationships_REL_2.csv...\n",
      "reading ./random/relationships_REL_0.csv...\n",
      "reading ./random/relationships_REL_3.csv...\n"
     ]
    }
   ],
   "source": [
    "import pyimport as pi\n",
    "\n",
    "node, rels = pi.load_dir('./random')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31182b70-aca6-4a8f-b56f-c48168e8ecb1",
   "metadata": {},
   "source": [
    "### At this point, we've got 2 PyArrow Tables!\n",
    "\n",
    "One caveat for now is the PyArrow CSV readers don't handle arrays. This can be worked around with Pandas + NumPy, but for this demo just pretend."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "5aa1e90f-a8c6-4f27-a09e-fd40fbea191d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ID: int64\n",
       "louvain: int64\n",
       "_labels_: list<item: string>\n",
       "  child 0, item: string"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "node.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "9879e4d1-7b31-4166-bacb-3f5776373ade",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "START_ID: int64\n",
       "END_ID: int64\n",
       "score: double\n",
       "_type_: string"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rels.schema"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "477e9d48-549e-443b-8e9d-e455199663aa",
   "metadata": {},
   "source": [
    "### Sending the Data\n",
    "\n",
    "We do a similar dance like with other `neo4j-arrow` routines:\n",
    "\n",
    "1. Define and create a new server-side Job\n",
    "2. Stream our Nodes Table to the server\n",
    "3. Strem our Relationships Table to the server\n",
    "\n",
    "The _NEW_ thing with this bulk import job is the concept of 1 Job but 2 Streams!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "22584b3e-0314-40c9-9cc2-bfe94e8ede69",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sending 10,000,000 nodes...\n",
      "Sending 140,159,916 relationships...\n",
      "Waiting for job completion...\n",
      "Done!\n"
     ]
    }
   ],
   "source": [
    "ticket = client.bulk_import(database='demotime123', \n",
    "                            id_field='ID', labels_field='_labels_',\n",
    "                            source_field='START_ID', target_field='END_ID', type_field='_type_')\n",
    "\n",
    "print(f'Sending {len(node):,} nodes...')\n",
    "client.put(ticket, node, metadata={'stream.type': 'node'})\n",
    "\n",
    "print(f'Sending {len(rels):,} relationships...')\n",
    "client.put(ticket, rels, metadata={'stream.type': 'rels'})\n",
    "\n",
    "print('Waiting for job completion...')\n",
    "client.wait_for_job(ticket, desired=na.JobStatus.COMPLETE, timeout=600)\n",
    "\n",
    "print('Done!')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3fce3006-9490-41ee-b6e8-7092a0fa9961",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46f1ae8b-3592-498b-8e40-cf4c6bd8c583",
   "metadata": {},
   "source": [
    "## Future Work\n",
    "\n",
    "1. The above is very unoptimized (single-threaded reads of CSVs, no tuning of batch sizes for streams, etc.)\n",
    "  - Concurrency _can_ be increased. Lowest hanging fruit is sending nodes and rels concurrently.\n",
    "  - CSVs suck...enough said.\n",
    "    \n",
    "2. No _incremental_ import\n",
    "  - PRD exists https://docs.google.com/document/d/1l3MbrrZlEG_2QRFAnNu76kBJzUZ3Z1GrkUPUmQjJqcE/edit#\n",
    "  - Thought experiment: _is it faster to build a new database and use Fabric?_\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "60db6ebb-65cd-46f3-97bf-eddaa750c8c6",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}