lineage.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 2,
  "metadata": {
    "kernelspec": {
      "name": "synapse_pyspark",
      "display_name": "Synapse PySpark"
    },
    "language_info": {
      "name": "python"
    },
    "description": null,
    "save_output": true,
    "synapse_widget": {
      "version": "0.1",
      "state": {}
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Overview\n",
        "\n",
        "This notebook uses pyapacheatlas to inject lineage between input and output assets "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "outputs": [],
      "metadata": {},
      "source": [
        "# Install Atlas Python client (https://github.com/wjohnson/pyapacheatlas)\n",
        "import pyapacheatlas\n",
        "import json"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "outputs": [],
      "metadata": {},
      "source": [
        "from pyapacheatlas.auth import ServicePrincipalAuthentication\n",
        "from pyapacheatlas.core import PurviewClient\n",
        "\n",
        "# enter your service principal credentials here\n",
        "oauth = ServicePrincipalAuthentication(\n",
        "    tenant_id= \"\",\n",
        "    client_id=\"\",\n",
        "    client_secret=\"\"\n",
        ")\n",
        ""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "outputs": [],
      "metadata": {},
      "source": [
        "# Instantiate PurviewClient, enter your Purview account name here\n",
        "client = PurviewClient(\n",
        "    account_name =  \"account-name-goes-here\",\n",
        "    authentication=oauth\n",
        ")"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "outputs": [],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        "\r\n",
        "from pyapacheatlas.core import AtlasEntity\r\n",
        "from pyapacheatlas.core import AtlasProcess\r\n",
        "\r\n",
        "import uuid\r\n",
        "\r\n",
        "# Specify input(s) the output(s), and the process\r\n",
        "\r\n",
        "# Define an existing entity or create a new entity\r\n",
        "# You must provide a name, typeName, qualified_name, and guid\r\n",
        "# the guid must be a negative number if it to be created\r\n",
        "\r\n",
        "vuuid = \"-\"+ str(uuid.uuid4())\r\n",
        "input01 = AtlasEntity(\r\n",
        "    name=\"alpha.json\",\r\n",
        "    typeName=\"azure_datalake_gen2_path\",\r\n",
        "    qualified_name=\"https://somedatalake.dfs.core.windows.net/raw/acme/datasets/alpha.json\",\r\n",
        "    guid=vuuid\r\n",
        ")\r\n",
        "vuuid = \"-\"+ str(uuid.uuid4())\r\n",
        "output01 = AtlasEntity(\r\n",
        "    name=\"alpha\",\r\n",
        "    typeName=\"azure_datalake_gen2_resource_set\",\r\n",
        "    qualified_name=\"https://somedatalake.dfs.core.windows.net/base/acme/alpha/{SparkPartitions}\",\r\n",
        "    guid=vuuid\r\n",
        ")\r\n",
        "\r\n",
        "# only need to run the next line if either of the assets do not exist yet.  So long as the qualified name is the same as the scanned asset Purview will update rather than create duplicate\r\n",
        "results = client.upload_entities(batch=[input01, output01])"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "outputs": [],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        "\r\n",
        "# The Atlas Process is the lineage component that links the two\r\n",
        "# entities together. The inputs and outputs need to be the \"header\"\r\n",
        "# version of the atlas entities, so specify minimum = True to\r\n",
        "# return just guid, qualifiedName, and typeName.\r\n",
        "\r\n",
        "process_qn = \"pyapacheatlas://acmebaseprocess\"\r\n",
        "process_type_name = \"Process\"\r\n",
        "vuuid = \"-\"+ str(uuid.uuid4())\r\n",
        "\r\n",
        "process = AtlasProcess(\r\n",
        "    name=\"Synapse Spark - process raw\",\r\n",
        "    attributes={\"description\":\"Spark job to transform raw files into standized format\"},\r\n",
        "    typeName=process_type_name,\r\n",
        "    qualified_name=process_qn,\r\n",
        "    inputs=[input01],\r\n",
        "    outputs=[output01],\r\n",
        "    guid=vuuid\r\n",
        ")\r\n",
        "\r\n",
        "# inject the following line after the attributes parameter above in order to assign experts and owners to the process\r\n",
        "#contacts={\"Expert\":[{\"id\":\"AAD OID\",\"info\":\"some additional info goes here\"}],\"Owner\":[{\"id\":\"AAD OID\",\"info\":\"some additional info goes here\"}]},    \r\n",
        "\r\n",
        "# Create the lineage component.\r\n",
        "results = client.upload_entities(\r\n",
        "    batch=[input01, output01, process]\r\n",
        ")\r\n",
        "\r\n",
        "print(json.dumps(results, indent=2))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "outputs": [],
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      },
      "source": [
        "\r\n",
        "from pyapacheatlas.core import AtlasEntity\r\n",
        "\r\n",
        "\r\n",
        "# Here is an example of creating lineage from multiple inputs\r\n",
        "\r\n",
        "input01 = AtlasEntity(\r\n",
        "    name=\"alpha\",\r\n",
        "    typeName=\"azure_datalake_gen2_resource_set\",\r\n",
        "    qualified_name=\"https://somedatalake.dfs.core.windows.net/base/acme/alpha/{SparkPartitions}\",\r\n",
        "    guid=\"9f398956-febc-4643-bc55-b32d9d5b5f21\"\r\n",
        ")\r\n",
        "input02 = AtlasEntity(\r\n",
        "    name=\"coffee\",\r\n",
        "    typeName=\"azure_datalake_gen2_resource_set\",\r\n",
        "    qualified_name=\"https://somedatalake.dfs.core.windows.net/base/contonso/coffee/{Department}/{SparkPartitions}\",\r\n",
        "    guid=\"9f398956-febc-4643-bc55-b32d9d5b6f21\"\r\n",
        ")\r\n",
        "output01 = AtlasEntity(\r\n",
        "    name=\"alphacoffee\",\r\n",
        "    typeName=\"azure_datalake_gen2_resource_set\",\r\n",
        "    qualified_name=\"https://somedatalake.dfs.core.windows.net/curated/alphacoffee/{Department}/{SparkPartitions}\",\r\n",
        "    guid=\"9f398956-febc-4643-bc55-b32d9d5b5f21\"\r\n",
        ")\r\n",
        "\r\n",
        "# only uncomment and run the next line if either of the assets do not exist yet. So long as the qualified name is the same as the scanned asset Purview will update rather than create duplicate\r\n",
        "#results = client.upload_entities(batch=[input01, input02, output01])"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "outputs": [],
      "metadata": {},
      "source": [
        "from pyapacheatlas.core import AtlasProcess\n",
        "\n",
        "\n",
        "# The Atlas Process is the lineage component that links the two\n",
        "# entities together. The inputs and outputs need to be the \"header\"\n",
        "# version of the atlas entities, so specify minimum = True to\n",
        "# return just guid, qualifiedName, and typeName.\n",
        "\n",
        "process_qn = \"pyapacheatlas://curationprocess\"\n",
        "process_type_name = \"Process\"\n",
        "\n",
        "process = AtlasProcess(\n",
        "    name=\"Synapse Spark process\",\n",
        "    typeName=process_type_name,\n",
        "    qualified_name=process_qn,\n",
        "    inputs=[input01, input02],\n",
        "    outputs=[output01],\n",
        "    guid=-403\n",
        ")\n",
        "\n",
        "# Create lineage\n",
        "results = client.upload_entities(\n",
        "    batch=[input01, input02, output01, process]\n",
        ")\n",
        "\n",
        "print(json.dumps(results, indent=2))"
      ]
    }
  ]
}