NPLv2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Set-up\n",
    "\n",
    "## Import Datalogue libraries\n",
    "\n",
    "Note, you'll need to have downloaded and installed the Datalogue SDK before this step will work.\n",
    "\n",
    "Right now, to do so you will need to get access through Artifactory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'0.28.3'"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Import Datalogue libraries \n",
    "from datalogue import *\n",
    "from datalogue.version import __version__\n",
    "from datalogue.models.ontology import *\n",
    "from datalogue.models.datastore_collection import *\n",
    "from datalogue.models.datastore import *\n",
    "from datalogue.models.datastore import GCSDatastoreDef \n",
    "from datalogue.models.credentials import *\n",
    "from datalogue.models.stream import *\n",
    "from datalogue.models.transformations import *\n",
    "from datalogue.models.transformations.structure import *\n",
    "from datalogue.dtl import Dtl, DtlCredentials\n",
    "from datalogue.models.training import DataRef\n",
    "\n",
    "# Import Datalogue Bag of Tricks\n",
    "from DTLBagOTricks import DTL as DTLHelper\n",
    "\n",
    "\n",
    "# Import other useful libraries\n",
    "from datetime import datetime, timedelta\n",
    "from os import environ\n",
    "import pandas\n",
    "from IPython.display import Image\n",
    "\n",
    "# Checks the version of the SDK is correct\n",
    "# The expected version is 0.28.3\n",
    "# If the SDK is not installed, run `! pip install datalogue` and restart the Jupyter Notebook kernel\n",
    "# If the wrong versions is installed, run `! pip install datalogue --upgrade` and restart the Jupyter Notebook kernel\n",
    "__version__"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Datalogue v0.28.3\n",
      "Logged in 'https://internal.dtl.systems/api' with 'chrisr@datalogue.io' account)\n"
     ]
    }
   ],
   "source": [
    "# Set host, username and password variables\n",
    "\n",
    "datalogue_host = \"https://internal.dtl.systems\"  # for connecting to internal (note)\n",
    "\n",
    "# datalogue_host = \"https://internal.dtl.systems\"  # for connecting to internal (note)\n",
    "# datalogue_host = \"http://10.2.161.119:3000\"  # for connecting to Eric's DGX\n",
    "email = \"\"\n",
    "password = \"\"\n",
    "\n",
    "# Log in to Datalogue\n",
    "BOT = DTLHelper(datalogue_host, email, password)\n",
    "dtl = BOT.dtl\n",
    "\n",
    "# Expected output Datalogue v0.28.3\n",
    "# \"Logged in '[host location]' with '[username]' account)\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Read Source Files from S3 bucket"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'list'>\n",
      "<class 'dict'>\n"
     ]
    }
   ],
   "source": [
    "from boto.s3.connection import S3Connection\n",
    "\n",
    "conn = S3Connection('AWS Access Key','AWS Secret Key')\n",
    "bucket = conn.get_bucket('bucket-name')\n",
    "\n",
    "keys = [\"store_name\", \"URL\"]\n",
    "npl_data = []\n",
    "\n",
    "for key in bucket.list():\n",
    "    if 'Loan' in key.name:\n",
    "        url=\"https://datalogue-demo.s3.amazonaws.com/\" + key.name\n",
    "        values = [\"demo-\"+key.name, url]\n",
    "        npl_data.append(dict(zip(keys, values)))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "CSV Customers Sources to connect to:\n",
      "-------------------------\n",
      "➜ demo-LoanTape1USD.csv\n",
      "➜ demo-LoanTape2USD.csv\n",
      "➜ demo-LoanTape3EUR.csv\n",
      "➜ demo-LoanTape4GBP.csv\n",
      "➜ demo-LoanTape5USD.csv\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "\n",
    "print(\"\\nCSV Customers Sources to connect to:\\n\" \"-------------------------\")\n",
    "for data_store in npl_data:\n",
    "    print(\"➜ \" + data_store[\"store_name\"])\n",
    "print(\"\\n\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Create datastore connections for each file in S3 bucket"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'list'>\n",
      "<class 'datalogue.models.datastore.Datastore'>\n"
     ]
    }
   ],
   "source": [
    "current_stores = []\n",
    "\n",
    "for data_store in npl_data:\n",
    "    data_store[\"datastore_object\"] = dtl.datastore.create(\n",
    "        Datastore(\n",
    "            data_store[\"store_name\"],\n",
    "            HttpDatastoreDef(data_store[\"URL\"], FileFormat.Csv),\n",
    "        )\n",
    "    )\n",
    "    current_stores.append(data_store[\"datastore_object\"])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###           3b. Create datastore for RDBMS target"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "# host: 34.74.11.127 (use jdbc:postgresql://34.74.11.127:5432/demo for creating target store)\n",
    "# user: postgres\n",
    "# pw: L8am0pO5zjJrFm2O\n",
    "\n",
    "# bug in SDK for v<1.0; to be updated here but created in GUI for now\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Collecting data stores into a collection\n",
    "\n",
    "This is just used for organization, and uses the command `dtl.datastore_collection.create`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [],
   "source": [
    "npl_collection = DatastoreCollection(\n",
    "  name =\"collectionName\",\n",
    "  storeIds = [Datastore[\"datastore_object\"].id for Datastore in npl_data],\n",
    "  description = \"NPL tape data of various formats\"\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "npl_collection2 = dtl.datastore_collection.create(npl_collection)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5bb0253e-a184-4fd6-ab6b-2dd507b14374 | demo-NPL Collection |\n",
      "Datastores in Dataset:\n",
      "* demo-LoanTape2USD.csv --> 0c7d336e-a625-4bf1-906e-b3f66a7cca09\n",
      "* demo-LoanTape1USD.csv --> 51223aad-12ff-4336-89a6-2eb5cd35c272\n",
      "* demo-LoanTape3EUR.csv --> a122e970-70b8-4fc2-8406-a8d80982e4ef\n",
      "------------------------------------------------------------------------------------\n",
      "3e7a2dfa-6a94-42a3-a86f-7021f89d60d8 | demo-NPL Collection |\n",
      "Datastores in Dataset:\n",
      "* demo-LoanTape3EUR.csv --> 50d64640-2251-4432-92e3-d5ead126fe7d\n",
      "* demo-LoanTape1USD.csv --> d15bc2b0-2cd2-45ea-8b6c-50cb22c8367a\n",
      "* demo-LoanTape2USD.csv --> ea0e139a-fdfe-409f-9fd6-273f0c4bc6a4\n",
      "------------------------------------------------------------------------------------\n",
      "77c60112-b491-4901-a584-30e036ae29e8 | demo-NPL Collection |\n",
      "Datastores in Dataset:\n",
      "* demo-LoanTape2USD.csv --> 3a565393-2844-47d0-be54-b9a5fee9ada2\n",
      "* demo-LoanTape5USD.csv --> 9b7f5b18-6b83-4e9c-8e62-461b59c4456c\n",
      "* demo-LoanTape4GBP.csv --> decfb804-32c3-4462-ba3a-346bf1a5bd8e\n",
      "* demo-LoanTape3EUR.csv --> e2e4e474-f675-4624-977e-28257e83c45d\n",
      "* demo-LoanTape1USD.csv --> eceb8dae-4885-40c0-b716-11413cbd21ac\n",
      "------------------------------------------------------------------------------------\n",
      "068ae8cc-6241-44dc-8267-08c68e1e26fa | demo-NPL Collection |\n",
      "Datastores in Dataset:\n",
      "* demo-LoanTape5USD.csv --> 3507f986-03f5-4900-bfc4-2f368b13c16d\n",
      "* demo-LoanTape1USD.csv --> 5bd45b9c-f541-4372-baba-bfebb63e0542\n",
      "* demo-LoanTape4GBP.csv --> cb2cd9c6-f965-4985-9bd3-c8c6697a49ae\n",
      "* demo-LoanTape2USD.csv --> f2609dc2-dfc5-4600-93d4-4aa5334184bf\n",
      "* demo-LoanTape3EUR.csv --> f9296f81-97fd-48bd-822a-cd8457d32402\n",
      "------------------------------------------------------------------------------------\n",
      "7e9b6b3a-71a2-469b-a582-6277c89c57a6 | demo-NPL Collection |\n",
      "Datastores in Dataset:\n",
      "* demo-LoanTape1USD.csv --> 0a0118b5-98b7-4ec0-8b0f-2ec0ad76b9b7\n",
      "* demo-LoanTape3EUR.csv --> 7253975b-1f8e-4aac-9d90-6c7f3d7642e1\n",
      "* demo-LoanTape5USD.csv --> d7674c92-cdfa-4172-ad4f-99a82244394f\n",
      "* demo-LoanTape4GBP.csv --> db111117-04be-4022-9ece-8054f9951b7d\n",
      "* demo-LoanTape2USD.csv --> ef158eac-da79-4d11-9dac-c7293810bb88\n",
      "------------------------------------------------------------------------------------\n",
      "input store id list: ['0c7d336e-a625-4bf1-906e-b3f66a7cca09', '51223aad-12ff-4336-89a6-2eb5cd35c272', 'a122e970-70b8-4fc2-8406-a8d80982e4ef', '50d64640-2251-4432-92e3-d5ead126fe7d', 'd15bc2b0-2cd2-45ea-8b6c-50cb22c8367a', 'ea0e139a-fdfe-409f-9fd6-273f0c4bc6a4', '3a565393-2844-47d0-be54-b9a5fee9ada2', '9b7f5b18-6b83-4e9c-8e62-461b59c4456c', 'decfb804-32c3-4462-ba3a-346bf1a5bd8e', 'e2e4e474-f675-4624-977e-28257e83c45d', 'eceb8dae-4885-40c0-b716-11413cbd21ac', '3507f986-03f5-4900-bfc4-2f368b13c16d', '5bd45b9c-f541-4372-baba-bfebb63e0542', 'cb2cd9c6-f965-4985-9bd3-c8c6697a49ae', 'f2609dc2-dfc5-4600-93d4-4aa5334184bf', 'f9296f81-97fd-48bd-822a-cd8457d32402', '0a0118b5-98b7-4ec0-8b0f-2ec0ad76b9b7', '7253975b-1f8e-4aac-9d90-6c7f3d7642e1', 'd7674c92-cdfa-4172-ad4f-99a82244394f', 'db111117-04be-4022-9ece-8054f9951b7d', 'ef158eac-da79-4d11-9dac-c7293810bb88']\n"
     ]
    }
   ],
   "source": [
    "# Checking that the collection was created\n",
    "\n",
    "input_store_list = []\n",
    "\n",
    "for dataset in dtl.datastore_collection.list():\n",
    "    if (dataset.name == \"collectionName\"):\n",
    "        print(dataset.id, \"|\", dataset.name, \"|\")\n",
    "        print(\"Datastores in Dataset:\")\n",
    "        for datastore in dataset.storeIds:\n",
    "            print(\"* \" + datastore[\"name\"] + \" --> \" + datastore[\"id\"])\n",
    "            input_store_list.append(datastore[\"id\"])\n",
    "        print(\n",
    "            \"------------------------------------------------------------------------------------\"\n",
    "        )\n",
    "\n",
    "print('input store id list: ' + str(input_store_list))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Creating a stream\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check if any existing now \n",
    "#for stream in dtl.stream_collection.list():\n",
    "#    print('Stream name: ', stream.name )\n",
    "#    print('Stream ID: ', stream.id)\n",
    "#print('-----------------------')\n",
    "#print('Total of ', len(dtl.stream_collection.list()), ' stream(s).')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_output_store = dtl.datastore.get(\"5d4f4f1b-55d6-4ef0-823e-d37d7f58dc3f\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Sample pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define the target output schema transformation using 'structure'\n",
    "\n",
    "std_schema = Structure([\n",
    "        ClassNodeDescription(\n",
    "            path = [\"LoanID\"],\n",
    "            tag = \"Loan ID\",\n",
    "            pick_strategy = PickStrategy.HighScore,\n",
    "            data_type = DataType.String\n",
    "        ),\n",
    "        ClassNodeDescription(\n",
    "            path = [\"Unpaid_Principal\"],\n",
    "            tag = \"Unpaid Principal Balance\",\n",
    "            pick_strategy = PickStrategy.HighScore,\n",
    "            data_type = DataType.String\n",
    "        ),\n",
    "        ClassNodeDescription(\n",
    "            path = [\"Orig_Val\"],\n",
    "            tag = \"Origination Value\",\n",
    "            pick_strategy = PickStrategy.HighScore,\n",
    "            data_type = DataType.String\n",
    "        ),\n",
    "        ClassNodeDescription(\n",
    "            path = [\"Int_Type\"],\n",
    "            tag = \"Interest Type\",\n",
    "            pick_strategy = PickStrategy.HighScore,\n",
    "            data_type = DataType.String\n",
    "        ),\n",
    "        ClassNodeDescription(\n",
    "            path = [\"Mat_Date\"],\n",
    "            tag = \"Maturity Date\",\n",
    "            pick_strategy = PickStrategy.HighScore,\n",
    "            data_type = DataType.String\n",
    "        ),\n",
    "        ClassNodeDescription(\n",
    "            path = [\"Currency\"],\n",
    "            tag = \"Currency\",\n",
    "            pick_strategy = PickStrategy.HighScore,\n",
    "            data_type = DataType.String\n",
    "        ),\n",
    "        ClassNodeDescription(\n",
    "            path = [\"Orig_Country\"],\n",
    "            tag = \"Country of Origination\",\n",
    "            pick_strategy = PickStrategy.HighScore,\n",
    "            data_type = DataType.String\n",
    "        ),\n",
    "        ClassNodeDescription(\n",
    "            path = [\"Days_Past_Due\"],\n",
    "            tag = \"Days Past Due\",\n",
    "            pick_strategy = PickStrategy.HighScore,\n",
    "            data_type = DataType.String\n",
    "        )\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define classify transformation\n",
    "\n",
    "from datalogue.models.transformations import ReplaceLabel\n",
    "\n",
    "tx_definition = Definition(    # (List[Transformation], pipelines: List['Definition'], target_datastore )\n",
    "            [\n",
    "                Classify(use_context=True, include_classes=False, include_scores=False),\n",
    "                std_schema\n",
    "            ], # List of transformations\n",
    "            [], # pipelines list\n",
    "            my_output_store, # target_datastore\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'datalogue.models.stream.Definition'>\n"
     ]
    }
   ],
   "source": [
    "print(type(tx_definition))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'list'>\n",
      "<class 'datalogue.models.stream.Stream'>\n"
     ]
    }
   ],
   "source": [
    "# Define n stream(s), where n is number of datastore connections created from S3 bucket scan\n",
    "n = len(current_stores)\n",
    "i = 1\n",
    "\n",
    "list_of_streams = []\n",
    "for i in range(n):\n",
    "    stream = Stream(current_stores[i], [tx_definition])\n",
    "    i += 1\n",
    "    list_of_streams.append(stream)\n",
    "\n",
    "print(type(list_of_streams))\n",
    "print(type(list_of_streams[0]))    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "StreamCollection(id: b8156451-8045-4d6c-a863-ae3d2ff32782, name: 'NPLDemoPipeline', streams: [StreamMetadata(id: 776f90e0-9d7b-4246-86c0-13c234f9a4c6, is_ready: True, stream: Stream(type: <datalogue.models.datastore.HttpDatastoreDef object at 0x0000024A1D7B2E80>, pipelines: [Pipeline(type: [Classify(paths: , options: UseContext), Structure(structure: [ClassNodeDescription(path: ['LoanID'], tag: Loan ID, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Unpaid_Principal'], tag: Unpaid Principal Balance, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Val'], tag: Origination Value, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Int_Type'], tag: Interest Type, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Mat_Date'], tag: Maturity Date, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Currency'], tag: Currency, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Country'], tag: Country of Origination, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Days_Past_Due'], tag: Days Past Due, strategy: PickStrategy.HighScore, dataType: DataType.String)])], pipelines: [], target: <datalogue.models.datastore.JdbcDatastoreDef object at 0x0000024A1D82C518>)], env: None)), StreamMetadata(id: 213558fd-13d3-46ba-a6e3-20d22d39eadd, is_ready: True, stream: Stream(type: <datalogue.models.datastore.HttpDatastoreDef object at 0x0000024A1D800E80>, pipelines: [Pipeline(type: [Classify(paths: , options: UseContext), Structure(structure: [ClassNodeDescription(path: ['LoanID'], tag: Loan ID, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Unpaid_Principal'], tag: Unpaid Principal Balance, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Val'], tag: Origination Value, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Int_Type'], tag: Interest Type, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Mat_Date'], tag: Maturity Date, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Currency'], tag: Currency, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Country'], tag: Country of Origination, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Days_Past_Due'], tag: Days Past Due, strategy: PickStrategy.HighScore, dataType: DataType.String)])], pipelines: [], target: <datalogue.models.datastore.JdbcDatastoreDef object at 0x0000024A1D800CF8>)], env: None))]) \n",
      "\n",
      "StreamCollection(id: 20291b92-ffce-4da4-8c10-609de54d9376, name: 'NPLDemoPipeline', streams: [StreamMetadata(id: 9622aa59-6fae-4a57-bd97-67888636f773, is_ready: True, stream: Stream(type: <datalogue.models.datastore.HttpDatastoreDef object at 0x0000024A1D866D30>, pipelines: [Pipeline(type: [Classify(paths: , options: UseContext), Structure(structure: [ClassNodeDescription(path: ['LoanID'], tag: Loan ID, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Unpaid_Principal'], tag: Unpaid Principal Balance, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Val'], tag: Origination Value, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Int_Type'], tag: Interest Type, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Mat_Date'], tag: Maturity Date, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Currency'], tag: Currency, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Country'], tag: Country of Origination, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Days_Past_Due'], tag: Days Past Due, strategy: PickStrategy.HighScore, dataType: DataType.String)])], pipelines: [], target: <datalogue.models.datastore.JdbcDatastoreDef object at 0x0000024A1D866DA0>)], env: None)), StreamMetadata(id: 79b032a6-d8c1-42ad-a750-1e8e2a588834, is_ready: True, stream: Stream(type: <datalogue.models.datastore.HttpDatastoreDef object at 0x0000024A1D7BC0F0>, pipelines: [Pipeline(type: [Classify(paths: , options: UseContext), Structure(structure: [ClassNodeDescription(path: ['LoanID'], tag: Loan ID, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Unpaid_Principal'], tag: Unpaid Principal Balance, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Val'], tag: Origination Value, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Int_Type'], tag: Interest Type, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Mat_Date'], tag: Maturity Date, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Currency'], tag: Currency, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Country'], tag: Country of Origination, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Days_Past_Due'], tag: Days Past Due, strategy: PickStrategy.HighScore, dataType: DataType.String)])], pipelines: [], target: <datalogue.models.datastore.JdbcDatastoreDef object at 0x0000024A1D7BC160>)], env: None))]) \n",
      "\n",
      "StreamCollection(id: ac20d796-dbdd-4855-9f99-a09a89f080c8, name: 'NPLDemoPipeline', streams: [StreamMetadata(id: 4bc32f47-7934-4897-af91-cd149655fda7, is_ready: True, stream: Stream(type: <datalogue.models.datastore.HttpDatastoreDef object at 0x0000024A1D7C6898>, pipelines: [Pipeline(type: [Classify(paths: , options: UseContext), Structure(structure: [ClassNodeDescription(path: ['LoanID'], tag: Loan ID, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Unpaid_Principal'], tag: Unpaid Principal Balance, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Val'], tag: Origination Value, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Int_Type'], tag: Interest Type, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Mat_Date'], tag: Maturity Date, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Currency'], tag: Currency, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Country'], tag: Country of Origination, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Days_Past_Due'], tag: Days Past Due, strategy: PickStrategy.HighScore, dataType: DataType.String)])], pipelines: [], target: <datalogue.models.datastore.JdbcDatastoreDef object at 0x0000024A1D7C6908>)], env: None)), StreamMetadata(id: 228af0d4-3fc2-4d0e-b16d-b3a018c6c7ab, is_ready: True, stream: Stream(type: <datalogue.models.datastore.HttpDatastoreDef object at 0x0000024A1D7C6C18>, pipelines: [Pipeline(type: [Classify(paths: , options: UseContext), Structure(structure: [ClassNodeDescription(path: ['LoanID'], tag: Loan ID, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Unpaid_Principal'], tag: Unpaid Principal Balance, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Val'], tag: Origination Value, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Int_Type'], tag: Interest Type, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Mat_Date'], tag: Maturity Date, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Currency'], tag: Currency, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Country'], tag: Country of Origination, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Days_Past_Due'], tag: Days Past Due, strategy: PickStrategy.HighScore, dataType: DataType.String)])], pipelines: [], target: <datalogue.models.datastore.JdbcDatastoreDef object at 0x0000024A1D7C6C88>)], env: None))]) \n",
      "\n",
      "StreamCollection(id: d1385008-5a62-4e49-96f8-00d3b934c081, name: 'NPLDemoPipeline', streams: [StreamMetadata(id: b488e742-0a55-4132-b7d5-85ed590fb619, is_ready: True, stream: Stream(type: <datalogue.models.datastore.HttpDatastoreDef object at 0x0000024A1D8A1D68>, pipelines: [Pipeline(type: [Classify(paths: , options: UseContext), Structure(structure: [ClassNodeDescription(path: ['LoanID'], tag: Loan ID, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Unpaid_Principal'], tag: Unpaid Principal Balance, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Val'], tag: Origination Value, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Int_Type'], tag: Interest Type, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Mat_Date'], tag: Maturity Date, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Currency'], tag: Currency, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Country'], tag: Country of Origination, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Days_Past_Due'], tag: Days Past Due, strategy: PickStrategy.HighScore, dataType: DataType.String)])], pipelines: [], target: <datalogue.models.datastore.JdbcDatastoreDef object at 0x0000024A1D8A1DD8>)], env: None)), StreamMetadata(id: 1c693ffc-a07a-42cb-8035-606f4063791a, is_ready: True, stream: Stream(type: <datalogue.models.datastore.HttpDatastoreDef object at 0x0000024A1D889128>, pipelines: [Pipeline(type: [Classify(paths: , options: UseContext), Structure(structure: [ClassNodeDescription(path: ['LoanID'], tag: Loan ID, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Unpaid_Principal'], tag: Unpaid Principal Balance, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Val'], tag: Origination Value, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Int_Type'], tag: Interest Type, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Mat_Date'], tag: Maturity Date, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Currency'], tag: Currency, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Orig_Country'], tag: Country of Origination, strategy: PickStrategy.HighScore, dataType: DataType.String), ClassNodeDescription(path: ['Days_Past_Due'], tag: Days Past Due, strategy: PickStrategy.HighScore, dataType: DataType.String)])], pipelines: [], target: <datalogue.models.datastore.JdbcDatastoreDef object at 0x0000024A1D889198>)], env: None))]) \n",
      "\n",
      "deleting stream ID:  d1385008-5a62-4e49-96f8-00d3b934c081\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# search the existing stream collections\n",
    "x=dtl.stream_collection.list()\n",
    "scid = ''\n",
    "for i in x:\n",
    "    if i.name == 'NPLDemoPipeline':\n",
    "        scid = i.id\n",
    "\n",
    "# delete the existing stream collection\n",
    "dtl.stream_collection.delete(stream_collection_id = scid)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'list'>\n",
      "<class 'datalogue.models.stream.Stream'>\n"
     ]
    }
   ],
   "source": [
    "# Put the streams in a collection\n",
    "\n",
    "stream_collection2 = dtl.stream_collection.create(\n",
    "    list_of_streams,\n",
    "    \"StreamCollectionName\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Run the Collection\n",
    "\n",
    "dtl.stream_collection.run(stream_collection2.id)"
   ]
  }
 ],
 "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.6.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}