ml-insights/sample_notebooks/4_Data_Source_and_Reader_Example.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ML Insights Data Reader & Data Source Example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Use Case\n",
    "\n",
    "This Notebook shows example of different Data Reader & Data Source for reading data for a specific format (CSV, JSON, JSONL data types).\n",
    "\n",
    "### About Dataset\n",
    "The Iris flower data set or Fisher's Iris data set is a multivariate data set . The data set consists of 50 samples from each of three species of Iris (Iris setosa, Iris virginica and Iris versicolor). Four features were measured from each sample: the length and the width of the sepals and petals, in centimeters.\n",
    "\n",
    "Dataset source : https://archive.ics.uci.edu/dataset/53/iris"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Install ML Observability Insights Library SDK\n",
    "\n",
    "- Prerequisites\n",
    "    - Linux/Mac (Intel CPU)\n",
    "    - Python 3.8 and 3.9 only\n",
    "\n",
    "\n",
    "- Installation\n",
    "    - ML Insights is made available as a Python package (via Artifactory) which can be installed using pip install as shown below. Depending on the execution engine on which to do the run, one can use scoped package. For eg: if we want to run on dask, use oracle-ml-insights[dask], for spark use oracle-ml-insights[spark], for native use oracle-ml-insights. One can install all the dependencies as use oracle-ml-insights[all]\n",
    "\n",
    "      !pip install oracle-ml-insights\n",
    "\n",
    "Refer : [Installation and Setup](https://docs.oracle.com/en-us/iaas/tools/ml-insights-docs/latest/ml-insights-documentation/html/user_guide/tutorials/install.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!python3 -m pip install oracle-ml-insights"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1 ML Insights Imports "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# imports\n",
    "\n",
    "import os\n",
    "from typing import Any\n",
    "import pyarrow as pa\n",
    "import pandas as pd\n",
    "\n",
    "# Import metrics\n",
    "from mlm_insights.core.features.feature import FeatureMetadata\n",
    "from mlm_insights.core.metrics.max import Max\n",
    "from mlm_insights.core.metrics.min import Min\n",
    "from mlm_insights.core.metrics.rows_count import RowCount\n",
    "from mlm_insights.core.metrics.quartiles import Quartiles\n",
    "from mlm_insights.core.metrics.sum import Sum\n",
    "\n",
    "from mlm_insights.builder.builder_component import MetricDetail, EngineDetail\n",
    "from mlm_insights.constants.types import FeatureType, DataType, VariableType\n",
    "from mlm_insights.core.metrics.metric_metadata import MetricMetadata\n",
    "\n",
    "# import data reader\n",
    "from mlm_insights.core.data_sources import LocalDatePrefixDataSource\n",
    "from mlm_insights.core.data_sources.oci_object_storage_file_search_data_source import ObjectStorageFileSearchDataSource\n",
    "from mlm_insights.core.data_sources import OCIDatePrefixDataSource\n",
    "from mlm_insights.mlm_native.readers import CSVNativeDataReader, NestedJsonNativeDataReader\n",
    "\n",
    "from mlm_insights.builder.insights_builder import InsightsBuilder\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2 Configure Feature schema\n",
    "\n",
    "Feature Schema defines the structure and metadata of the input data, which includes data type, column type, column mapping . The framework, uses this information as the ground truth and any deviation in the actual data is taken as an anomaly and the framework usually will ignore such all such anomaly in data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_input_schema():\n",
    "    return {\n",
    "        \"sepal length (cm)\": FeatureType(data_type=DataType.FLOAT, variable_type=VariableType.CONTINUOUS),\n",
    "        \"sepal width (cm)\": FeatureType(data_type=DataType.FLOAT, variable_type=VariableType.CONTINUOUS),\n",
    "        \"petal length (cm)\": FeatureType(data_type=DataType.FLOAT, variable_type=VariableType.CONTINUOUS),\n",
    "        \"petal width (cm)\": FeatureType(data_type=DataType.FLOAT, variable_type=VariableType.CONTINUOUS)\n",
    "    }\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3 Configure Metrics\n",
    "\n",
    "Metrics are the core construct for the framework. This component is responsible for calculating all statistical metrics and algorithms. Metric components work based on the type of features (eg. input feature, output feature etc.) available, their data type (eg. int, float, string etc.) as well as additional context (e.g. if any previous computation is available to compare against). ML Insights provides commonly used metrics out of the box for different ML observability use cases.\n",
    "\n",
    "Refer : [Metrics Component Documentation](https://docs.oracle.com/en-us/iaas/tools/ml-insights-docs/latest/ml-insights-documentation/html/user_guide/getting_started/metrics_component.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_metrics():\n",
    "    metrics = [\n",
    "               MetricMetadata(klass=Sum),\n",
    "               MetricMetadata(klass=Quartiles),\n",
    "               MetricMetadata(klass=Max),\n",
    "               MetricMetadata(klass=Min)\n",
    "              ]\n",
    "    uni_variate_metrics = {\n",
    "        \"sepal length (cm)\": metrics,\n",
    "        \"sepal width (cm)\": metrics,\n",
    "        \"petal length (cm)\": metrics,\n",
    "        \"petal width (cm)\": metrics\n",
    "    }\n",
    "    \n",
    "    dataset_metrics = [MetricMetadata(klass=RowCount)]\n",
    "    metric_details = MetricDetail(univariate_metric=uni_variate_metrics,\n",
    "                                  dataset_metrics=dataset_metrics)\n",
    "    return metric_details"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4 Configure Data Reader\n",
    "\n",
    "Data Reader allows for ingestion of raw data into the framework. This component is primarily responsible for understanding different formats of data (e.g. jsonl, csv) etc. and how to properly read them. At its essence, the primary responsibility of this component is that given a set of valid file locations which represents file of a specific type, reader can properly decode the content and load them in memory.\n",
    "\n",
    "Additionally, Data Source component is an optional subcomponent, which is usually used along side the Reader. The primary responsibility of the data source component is to embed logic on filtering and partitioning of files to be read by the framework.\n",
    "\n",
    "Refer : [Data Reader Documentation](https://docs.oracle.com/en-us/iaas/tools/ml-insights-docs/latest/ml-insights-documentation/html/user_guide/getting_started/data_reader_component.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.1.1 Local File Data source and CSV Data Reader\n",
    "\n",
    "Below example shows how to list and load csv files present in the local system\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_reader():\n",
    "    data = {\n",
    "        \"file_type\": \"csv\",\n",
    "        \"date_range\": {\"start\": \"2023-06-24\", \"end\": \"2023-06-27\"}\n",
    "    }\n",
    "    base_location =\"input_data/iris_dataset\"\n",
    "    ds = LocalDatePrefixDataSource(base_location, **data)\n",
    "    print(ds.get_data_location())\n",
    "    csv_reader = CSVNativeDataReader(data_source=ds)\n",
    "    return csv_reader"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Compute the Profile \n",
    "\n",
    "Create the builder object which provides core set of api, using which user can set the behavior of their monitoring. By selecting what components and variants to run all aspects of the monitoring task can be customised and configured. \n",
    "\n",
    "The run() method is responsible to run the internal workflow. It also handles the life cycle of each component passed, which includes creation (if required), invoking interface functions, destroying etc . Additionally, runner also handles some more advanced operations like thread pooling, compute engine abstraction etc.\n",
    "\n",
    "Refer : [Builder Object Documentation](https://docs.oracle.com/en-us/iaas/tools/ml-insights-docs/latest/ml-insights-documentation/html/user_guide/getting_started/builder_object.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def main():    \n",
    "    # Set up the insights builder by passing: input schema, metric, data frame and engine details\n",
    "    runner = InsightsBuilder(). \\\n",
    "        with_input_schema(get_input_schema()). \\\n",
    "        with_metrics(metrics=get_metrics()). \\\n",
    "        with_reader(reader=get_reader()). \\\n",
    "        build()\n",
    "    # Other Insights components that can be configured are:\n",
    "    # - Custom Transformers (ability to transform incoming data frame to add/update/merge/delete/normalize etc features)\n",
    "    # - Conditional Features (ability to create new features from existing features using python expressions)\n",
    "    # - Tags (ability to provide custom metadata to be added as key-value pairs to a Profile)\n",
    "\n",
    "    # Run the evaluation\n",
    "    run_result = runner.run()\n",
    "    return run_result.profile\n",
    "    \n",
    "profile = main()\n",
    "profile.to_pandas()\n",
    "\n",
    "profile_json = profile.to_json()\n",
    "dataset_metrics = profile_json\n",
    "pd.json_normalize(dataset_metrics).T.dropna()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.1.2 Local File Data source and Nested Json Data Reader\n",
    "\n",
    "Below example shows how to list and load data from nested json files present in the local system"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_nested_json_data_reader():\n",
    "    data = {\n",
    "        \"file_type\": \"json\",\n",
    "        \"date_range\": {\"start\": \"2023-06-24\", \"end\": \"2023-06-27\"}\n",
    "    }\n",
    "    base_location =\"input_data/nested_json\"\n",
    "\n",
    "    query = \"iris_dataset[].[{column: 'sepal length (cm)', value: sepal_length}, {column: 'sepal width (cm)', value: sepal_width}, {column : 'petal length (cm)', value : petal_length}, {column : 'petal width (cm)', value : petal_width} ]\"\n",
    "    \n",
    "    ds = LocalDatePrefixDataSource(base_location, **data)\n",
    "    print(ds.get_data_location())\n",
    "    nested_json_data_reader = NestedJsonNativeDataReader(data_source=ds, query=query, query_engine_name= \"JMESPATH\")\n",
    "    #csv_reader = CSVNativeDataReader(file_path=base_location)\n",
    "    return nested_json_data_reader"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compute Profile using Nested Json Data Reader\n",
    "\n",
    "Create the builder object which provides core set of api, using which user can set the behavior of their monitoring. By selecting what components and variants to run all aspects of the monitoring task can be customised and configured. \n",
    "\n",
    "The run() method is responsible to run the internal workflow. It also handles the life cycle of each component passed, which includes creation (if required), invoking interface functions, destroying etc . Additionally, runner also handles some more advanced operations like thread pooling, compute engine abstraction etc.\n",
    "\n",
    "Refer : [Builder Object Documentation](https://docs.oracle.com/en-us/iaas/tools/ml-insights-docs/latest/ml-insights-documentation/html/user_guide/getting_started/builder_object.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def main():    \n",
    "    # Set up the insights builder by passing: input schema, metric, data frame and engine details\n",
    "    runner = InsightsBuilder(). \\\n",
    "        with_input_schema(get_input_schema()). \\\n",
    "        with_metrics(metrics=get_metrics()). \\\n",
    "        with_reader(reader=get_nested_json_data_reader()). \\\n",
    "        build()\n",
    "    # Other Insights components that can be configured are:\n",
    "    # - Custom Transformers (ability to transform incoming data frame to add/update/merge/delete/normalize etc features)\n",
    "    # - Conditional Features (ability to create new features from existing features using python expressions)\n",
    "    # - Tags (ability to provide custom metadata to be added as key-value pairs to a Profile)\n",
    "\n",
    "    # Run the evaluation\n",
    "    run_result = runner.run()\n",
    "    return run_result.profile\n",
    "    \n",
    "profile = main()\n",
    "profile.to_pandas()\n",
    "\n",
    "profile_json = profile.to_json()\n",
    "dataset_metrics = profile_json\n",
    "pd.json_normalize(dataset_metrics).T.dropna()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.2.1 Object Storage Data source and CSV Data Reader\n",
    "\n",
    "Below example shows see how to list and  load nested json files present in the Object Storage\n",
    "\n",
    "Need to enable OCI_RESOURCE_PRINCIPAL authentication for target object storage bucket to run the following in local or in customer tenancy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_object_storage_reader():\n",
    "    data = {\n",
    "        \"file_type\": \"csv\",\n",
    "        \"date_range\": {\"start\": \"2023-06-24\", \"end\": \"2023-06-27\"},\n",
    "        \"bucket_name\": \"mlm-insights\",\n",
    "        \"namespace\" : \"bigdatadatasciencelarge\",\n",
    "        \"object_prefix\" : \"input/iris_dataset\"\n",
    "    }\n",
    "    #base_location =\"oci://mlm-insights/input/iris_dataset\"\n",
    "    ds = OCIDatePrefixDataSource(**data)\n",
    "    print(ds.get_data_location())\n",
    "    csv_reader = CSVNativeDataReader(data_source=ds)\n",
    "    #csv_reader = CSVNativeDataReader(file_path=base_location)\n",
    "    return csv_reader"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Compute the Profile \n",
    "\n",
    "Create the builder object which provides core set of api, using which user can set the behavior of their monitoring. By selecting what components and variants to run all aspects of the monitoring task can be customised and configured. \n",
    "\n",
    "The run() method is responsible to run the internal workflow. It also handles the life cycle of each component passed, which includes creation (if required), invoking interface functions, destroying etc . Additionally, runner also handles some more advanced operations like thread pooling, compute engine abstraction etc.\n",
    "\n",
    "Refer : [Builder Object Documentation](https://docs.oracle.com/en-us/iaas/tools/ml-insights-docs/latest/ml-insights-documentation/html/user_guide/getting_started/builder_object.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def main():    \n",
    "    # Set up the insights builder by passing: input schema, metric, data frame and engine details\n",
    "    runner = InsightsBuilder(). \\\n",
    "        with_input_schema(get_input_schema()). \\\n",
    "        with_metrics(metrics=get_metrics()). \\\n",
    "        with_reader(reader=get_object_storage_reader()). \\\n",
    "        build()\n",
    "    # Other Insights components that can be configured are:\n",
    "    # - Custom Transformers (ability to transform incoming data frame to add/update/merge/delete/normalize etc features)\n",
    "    # - Conditional Features (ability to create new features from existing features using python expressions)\n",
    "    # - Tags (ability to provide custom metadata to be added as key-value pairs to a Profile)\n",
    "\n",
    "    # Run the evaluation\n",
    "    run_result = runner.run()\n",
    "    return run_result.profile\n",
    "    \n",
    "profile = main()\n",
    "profile.to_pandas()\n",
    "\n",
    "profile_json = profile.to_json()\n",
    "dataset_metrics = profile_json\n",
    "pd.json_normalize(dataset_metrics).T.dropna()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.2.2 Object Storage Data source and Nested Json Data Reader\n",
    "\n",
    "Below example shows see how to list and load csv files present in the Object Storage\n",
    "\n",
    "Need to enable OCI_RESOURCE_PRINCIPAL authentication for target object storage bucket to run the following in local or in customer tenancy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_object_storage_nested_json_data_reader():\n",
    "    \n",
    "    data = {\n",
    "        \"file_type\": \"json\",\n",
    "        \"date_range\": {\"start\": \"2023-06-24\", \"end\": \"2023-06-27\"},\n",
    "        \"bucket_name\": \"mlm-insights\",\n",
    "        \"namespace\" : \"bigdatadatasciencelarge\",\n",
    "        \"object_prefix\" : \"input/nested_json\"\n",
    "    }\n",
    "   \n",
    "    query = \"iris_dataset[].[{column: 'sepal length (cm)', value: sepal_length}, {column: 'sepal width (cm)', value: sepal_width}, {column : 'petal length (cm)', value : petal_length}, {column : 'petal width (cm)', value : petal_width} ]\"\n",
    "\n",
    "    ds = OCIDatePrefixDataSource(**data)\n",
    "    print(ds.get_data_location())\n",
    "    nested_json_data_reader = NestedJsonNativeDataReader(data_source=ds, query=query, query_engine_name= \"JMESPATH\")\n",
    "    return nested_json_data_reader"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compute the Profile for Nested Json Data Reader\n",
    "\n",
    "Create the builder object which provides core set of api, using which user can set the behavior of their monitoring. By selecting what components and variants to run all aspects of the monitoring task can be customised and configured. \n",
    "\n",
    "The run() method is responsible to run the internal workflow. It also handles the life cycle of each component passed, which includes creation (if required), invoking interface functions, destroying etc . Additionally, runner also handles some more advanced operations like thread pooling, compute engine abstraction etc.\n",
    "\n",
    "Refer : [Builder Object Documentation](https://docs.oracle.com/en-us/iaas/tools/ml-insights-docs/latest/ml-insights-documentation/html/user_guide/getting_started/builder_object.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def main():    \n",
    "    # Set up the insights builder by passing: input schema, metric, data frame and engine details\n",
    "    runner = InsightsBuilder(). \\\n",
    "        with_input_schema(get_input_schema()). \\\n",
    "        with_metrics(metrics=get_metrics()). \\\n",
    "        with_reader(reader=get_object_storage_nested_json_data_reader()). \\\n",
    "        build()\n",
    "    # Other Insights components that can be configured are:\n",
    "    # - Custom Transformers (ability to transform incoming data frame to add/update/merge/delete/normalize etc features)\n",
    "    # - Conditional Features (ability to create new features from existing features using python expressions)\n",
    "    # - Tags (ability to provide custom metadata to be added as key-value pairs to a Profile)\n",
    "\n",
    "    # Run the evaluation\n",
    "    run_result = runner.run()\n",
    "    return run_result.profile\n",
    "    \n",
    "profile = main()\n",
    "profile.to_pandas()\n",
    "\n",
    "profile_json = profile.to_json()\n",
    "dataset_metrics = profile_json\n",
    "pd.json_normalize(dataset_metrics).T.dropna()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.3.1 OCI Object Storage File Search Data source and CSV Data Reader\n",
    "\n",
    "Below example shows how to list and load csv files present in the local system"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import oci\n",
    "def get_reader():\n",
    "    data = [\n",
    "        { \"contains\": \"iris_dataset\"},\n",
    "        { \"last_n_days\" : \"30\"},\n",
    "        # {\"file_extension\" : \"csv\"},\n",
    "        {\"suffix\": \"iris.csv\"},\n",
    "        {\"prefix\": \"ml-mon-poc@ociodscdev/input/\"} \n",
    "        # {\"date_range\": {\"start\": \"2024-01-01\", \"end\": \"2024-01-09\", \"date_format\" : \"yyyy-mm-dd\", \"search_type\" : \"metadata\"}}\n",
    "    ]\n",
    "    bucket_name = \"ml-mon-poc\"\n",
    "    namespace = \"ociodscdev\"\n",
    "    base_locations = ['oci://%s@%s/%s' % (bucket_name, namespace, \"input/\"), 'oci://%s@%s/%s' % (bucket_name, namespace, \"input_2/\") ]\n",
    "\n",
    "    config = oci.config.from_file()\n",
    "    token_file = config['security_token_file']\n",
    "    token = None\n",
    "    with open(token_file, 'r') as f:\n",
    "        token = f.read()\n",
    "\n",
    "    private_key = oci.signer.load_private_key_from_file(config['key_file'])\n",
    "    signer = oci.auth.signers.SecurityTokenSigner(token, private_key)\n",
    "    storage_options = {\"signer\": signer, \"region\": \"us-ashburn-1\"}\n",
    "\n",
    "    \n",
    "    base_locations = ['oci://%s@%s/%s' % (bucket_name, namespace, \"input/\"), 'oci://%s@%s/%s' % (bucket_name, namespace, \"input_2/\") ]\n",
    "\n",
    "    ds = ObjectStorageFileSearchDataSource(file_path=base_locations, storage_options=storage_options, filter_arg=data)\n",
    "    print(ds.get_data_location())\n",
    "    csv_reader = CSVNativeDataReader(data_source=ds, storage_options=storage_options)\n",
    "    return csv_reader"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def main():    \n",
    "    # Set up the insights builder by passing: input schema, metric, data frame and engine details\n",
    "    runner = InsightsBuilder(). \\\n",
    "        with_input_schema(get_input_schema()). \\\n",
    "        with_metrics(metrics=get_metrics()). \\\n",
    "        with_reader(reader=get_reader()). \\\n",
    "        build()\n",
    "\n",
    "    # Run the evaluation\n",
    "    run_result = runner.run()\n",
    "    return run_result.profile\n",
    "    \n",
    "profile = main()\n",
    "profile.to_pandas()\n",
    "\n",
    "profile_json = profile.to_json()\n",
    "dataset_metrics = profile_json\n",
    "pd.json_normalize(dataset_metrics).T.dropna()"
   ]
  }
 ],
 "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.16"
  },
  "vscode": {
   "interpreter": {
    "hash": "06b89612a5e9c675d881f7c391886fce9eabd2126328a7f9c136f634c360fd8c"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}