advanced/Entity Matching Pipelines with SDK-experimental demo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "0baefc29",
   "metadata": {},
   "source": [
    "# Create Entity Matching pipelines with SDK-experimental demo\n",
    "\n",
    "This notebook shows how to create entity matching pipelines. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bd0b776b",
   "metadata": {},
   "source": [
    "## Accessing CDF\n",
    "This tutorial assumes you have some basic knowledge of CDF and the Python SDK. If not, please follow the 'lab' tutorials first.\n",
    "\n",
    "For this tutorial the 'contextualization' tenant is used. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e0c1d491",
   "metadata": {},
   "source": [
    "## Import modules\n",
    "We need to import some Python modules in order to interact with CDF. We will use the Python SDK with Experimental Extensions, which we below refer to as a client. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ca8785c3",
   "metadata": {},
   "outputs": [],
   "source": [
    "from cognite.experimental import CogniteClient\n",
    "from cognite.experimental.data_classes import EntityMatchingPipeline\n",
    "from datetime import date\n",
    "from getpass import getpass\n",
    "import os\n",
    "import requests"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2329ec16",
   "metadata": {},
   "source": [
    "## Create a client\n",
    "\n",
    "When you create the CogniteClient below, getpass will ask for your API key in an extra password field. Simply paste ypu publicdata API-key and press shift+enter."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "04a8fda6",
   "metadata": {},
   "outputs": [],
   "source": [
    "project = \"contextualization\"\n",
    "api_key = getpass(\"Please enter API-KEY: \")\n",
    "client = CogniteClient(project=project, api_key=api_key, client_name=\"dshub\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aedc1903",
   "metadata": {},
   "source": [
    "## Create an entity matching pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "01d903f8",
   "metadata": {},
   "source": [
    "First define the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98381ef5",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_external_id = f\"my_ds_hub_pipeline_{date.today()}\"\n",
    "\n",
    "my_pipeline = EntityMatchingPipeline(name=\"my_ds_hub_pipeline\",\n",
    "                                     description=\"Test pipeline created using dshub tutorial\",\n",
    "                                     external_id=my_external_id,\n",
    "                                     sources= {'dataSetIds': [{'id': 4677214669402260}],\n",
    "                                               'resource': 'timeseries'},\n",
    "                                     targets= {'dataSetIds': [{'id': 1181171615083226}], \n",
    "                                                    'resource': 'assets'},\n",
    "                                     model_parameters =  {'featureType': 'bigram',\n",
    "                                                          'matchFields': [{'source': 'name', 'target': 'name'}]}\n",
    "                                    )   "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed460258",
   "metadata": {},
   "source": [
    "Then create the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "12680435",
   "metadata": {},
   "outputs": [],
   "source": [
    "res = client.entity_matching.pipelines.create(pipeline=my_pipeline)\n",
    "my_pipeline_id = res.dump()[\"id\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "278119cc",
   "metadata": {},
   "source": [
    "Retrive the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0d471665",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_pipeline = client.entity_matching.pipelines.retrieve(id=my_pipeline_id)\n",
    "my_pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ea7fa858",
   "metadata": {},
   "source": [
    "List all pipelines and find the one you created by filtering on the external_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5cbf6016",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_list =  client.entity_matching.pipelines.list(limit=None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6b5b5db",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_pipeline_id = [pipeline[\"id\"] for pipeline in pipeline_list.dump() \n",
    "                  if (\"external_id\" in pipeline.keys() and (my_external_id in pipeline[\"external_id\"]))][0]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f1b67f31",
   "metadata": {},
   "source": [
    "## Run the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "adc4dc32",
   "metadata": {},
   "outputs": [],
   "source": [
    "res = client.entity_matching.pipelines.run(external_id=my_external_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "564e132b",
   "metadata": {},
   "source": [
    "### Look at the results\n",
    "\n",
    "List all runs for the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "432198a7",
   "metadata": {},
   "outputs": [],
   "source": [
    "client.entity_matching.pipelines.runs.list(id=my_pipeline_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d4e9fab2",
   "metadata": {},
   "source": [
    "List only last run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c20f1a6e",
   "metadata": {},
   "outputs": [],
   "source": [
    "last_run = client.entity_matching.pipelines.runs.retrieve_latest(id=my_pipeline_id)\n",
    "last_run_id = last_run.dump()['job_id']\n",
    "last_run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "26590795",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Not able to retrive the results from run with SDK -> using API\n",
    "headers = {\n",
    "    'Content-Type': 'application/json', 'API-key': api_key\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9af35efb",
   "metadata": {},
   "outputs": [],
   "source": [
    "url = f\"https://api.cognitedata.com/api/playground/projects/{project}/context/entitymatching/pipelines/run/{last_run_id}\"\n",
    "\n",
    "matches = []\n",
    "generatedRules = []\n",
    "response_get_pipeline_run = requests.get(url=url, headers=headers)\n",
    "run_results = response_get_pipeline_run.json()\n",
    "if run_results[\"status\"] != \"Completed\":\n",
    "    print(f\"Pipeline run is not completed, status is {run_results['status']}. Run again to check the status again.\")\n",
    "else:\n",
    "    print(\"Pipeline run is completed\")\n",
    "    matches = run_results[\"matches\"]\n",
    "    generatedRules = run_results[\"generatedRules\"]   "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4c8e3c14",
   "metadata": {},
   "source": [
    "Look at the results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8aa7f58d",
   "metadata": {},
   "outputs": [],
   "source": [
    "matches"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ceeff04b",
   "metadata": {},
   "outputs": [],
   "source": [
    "generatedRules"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0629d677",
   "metadata": {},
   "source": [
    "## Update the pipeline with matches and rules"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2df4b478",
   "metadata": {},
   "source": [
    "We looked through the first five matches and can confirm that these are correct. -> We want to update the pipeline with these as confirmed matches."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fad72fd2",
   "metadata": {},
   "outputs": [],
   "source": [
    "confirmed_matches = [{\"sourceId\": match[\"source\"][\"id\"],\n",
    "                      \"targetId\": match[\"target\"][\"id\"]} for match in run_results[\"matches\"][0:5]]\n",
    "update_request_body = {\n",
    "  \"items\": [\n",
    "    {\n",
    "      \"update\": {\n",
    "        \"confirmedMatches\": {\n",
    "            \"set\": confirmed_matches\n",
    "        }\n",
    "      },\n",
    "      \"id\": my_pipeline_id\n",
    "    }\n",
    "  ]\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7417aec8",
   "metadata": {},
   "outputs": [],
   "source": [
    "url_update = f\"https://api.cognitedata.com/api/playground/projects/{project}/context/entitymatching/pipelines/update\"\n",
    "response_update = requests.post(url=url_update, headers=headers, json=update_request_body)\n",
    "response_update.json()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b19a0275",
   "metadata": {},
   "outputs": [],
   "source": [
    "# We also want to confirm the two first of the generated rules"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7be858d9",
   "metadata": {},
   "outputs": [],
   "source": [
    "confirmed_rules = [{\"extractors\":rule['extractors'], \n",
    "                    \"conditions\":rule['conditions'],\n",
    "                    \"priority\":rule['priority']} for rule in run_results[\"generatedRules\"][0:2]]\n",
    "update_request_body = {\n",
    "  \"items\": [\n",
    "    {\n",
    "      \"update\": {\n",
    "        \"rules\": {\n",
    "            \"set\": confirmed_rules\n",
    "        }\n",
    "      },\n",
    "      \"id\": my_pipeline_id\n",
    "    }\n",
    "  ]\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "46d16709",
   "metadata": {},
   "outputs": [],
   "source": [
    "response_update = requests.post(url=url_update, headers=headers, json=update_request_body)\n",
    "response_update.json()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "701be150",
   "metadata": {},
   "source": [
    "## Run the pipeline again and look at the new results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2215a625",
   "metadata": {},
   "outputs": [],
   "source": [
    "res = client.entity_matching.pipelines.run(external_id=my_external_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cd9ba527",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_run_id= res.dump()[\"job_id\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cdee2882",
   "metadata": {},
   "outputs": [],
   "source": [
    "url = f\"https://api.cognitedata.com/api/playground/projects/{project}/context/entitymatching/pipelines/run/{pipeline_run_id}\"\n",
    "\n",
    "matches = []\n",
    "response_get_pipeline_run = requests.get(url=url, headers=headers)\n",
    "run_results = response_get_pipeline_run.json()\n",
    "if run_results[\"status\"] != \"Completed\":\n",
    "    print(f\"Pipeline run is not completed, status is {run_results['status']}. Run again to check the status again.\")\n",
    "else:\n",
    "    print(\"Pipeline run is completed\")\n",
    "    matches = run_results[\"matches\"]\n",
    "    generatedRules = run_results[\"generatedRules\"] "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "496db22f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# The matches we confirmed above have now 'matchType'='previously-confirmed' and score = 1.\n",
    "# Matches created by one of the rules we confirmed have 'matchType': 'match-rules X' and score = 1.\n",
    "matches"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0217d38c",
   "metadata": {},
   "source": [
    "## Delete a pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d671a9d8",
   "metadata": {},
   "outputs": [],
   "source": [
    "client.entity_matching.pipelines.delete(external_id=my_external_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bf660f38",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}