<commit_before>{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TFX - Compile and Deploy the TFX Pipeline to KFP\n",
    "\n",
    "This Notebook helps you to compile the **TFX Pipeline** to a **KFP package**. This will creat an **Argo YAML** file in a **.tar.gz** package. We perform the following steps:\n",
    "1. Build a custom container image that include our modules\n",
    "2. Compile TFX Pipeline using CLI\n",
    "3. Deploy the compiled pipeline to KFP \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install -q -U kfp"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Set compile time variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "os.environ['PROJECT_ID'] = 'ksalama-cloudml' # Set your project.\n",
    "os.environ['BUCKET'] = 'ksalama-cloudml' # Set your bucket.\n",
    "os.environ['REGION'] = 'us-central1' # Set your region.\n",
    "os.environ['GKE_CLUSTER_NAME'] = 'ksalama-cloudml-dev' # Set your GKE cluster name.\n",
    "os.environ['GKE_CLUSTER_ZONE'] = 'europe-west2-a' # Set your GKE cluster zone.\n",
    "\n",
    "os.environ['IMAGE_NAME'] = 'tfx-ml'\n",
    "os.environ['TAG'] = 'tfx0.25.0'\n",
    "os.environ['ML_IMAGE_URI']=f'gcr.io/{os.environ.get(\"PROJECT_ID\")}/{os.environ.get(\"IMAGE_NAME\")}:{os.environ.get(\"TAG\")}'\n",
    "\n",
    "os.environ['NAMESPACE'] = 'kubeflow-pipelines'\n",
    "os.environ['ARTIFACT_STORE_URI'] = f'gs://{os.environ.get(\"BUCKET\")}/tfx_artifact_store'\n",
    "os.environ['GCS_STAGING_PATH'] = f'{os.environ.get(\"ARTIFACT_STORE_URI\")}/staging'\n",
    "\n",
    "os.environ['RUNTIME_VERSION'] = '2.2'\n",
    "os.environ['PYTHON_VERSION'] = '3.7'\n",
    "os.environ['BEAM_RUNNER'] = 'DirectRunner'\n",
    "os.environ['MODEL_REGISTRY_URI'] = f'{os.environ.get(\"ARTIFACT_STORE_URI\")}/model_registry'\n",
    "\n",
    "os.environ['PIPELINE_NAME'] = 'tfx_bqml_scann'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tfx_pipeline import config\n",
    "\n",
    "for key, value in config.__dict__.items():\n",
    "  if key.isupper(): print(f'{key}: {value}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run the Pipeline Locally using Beam Runner"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import kfp\n",
    "import tfx\n",
    "from tfx.orchestration.beam.beam_dag_runner import BeamDagRunner\n",
    "from tfx_pipeline import pipeline as pipeline_module\n",
    "import tensorflow as tf\n",
    "import ml_metadata as mlmd\n",
    "from ml_metadata.proto import metadata_store_pb2\n",
    "import logging\n",
    "\n",
    "logging.getLogger().setLevel(logging.INFO)\n",
    "\n",
    "print(\"TFX Version:\", tfx.__version__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_root = f'{config.ARTIFACT_STORE_URI}/{config.PIPELINE_NAME}_beamrunner'\n",
    "model_regisrty_uri = f'{config.MODEL_REGISTRY_URI}_beamrunner'\n",
    "local_mlmd_sqllite = 'mlmd/mlmd.sqllite'\n",
    "\n",
    "print(f'Pipeline artifacts root: {pipeline_root}')    \n",
    "print(f'Model registry location: {model_regisrty_uri}')  \n",
    "\n",
    "if tf.io.gfile.exists(pipeline_root):\n",
    "  print(\"Removing previous artifacts...\")\n",
    "  tf.io.gfile.rmtree(pipeline_root)\n",
    "if tf.io.gfile.exists('mlmd'):\n",
    "  print(\"Removing local mlmd SQLite...\")\n",
    "  tf.io.gfile.rmtree('mlmd')\n",
    "print(\"Creating mlmd directory...\")\n",
    "tf.io.gfile.mkdir('mlmd')\n",
    "\n",
    "metadata_connection_config = metadata_store_pb2.ConnectionConfig()\n",
    "metadata_connection_config.sqlite.filename_uri = local_mlmd_sqllite\n",
    "metadata_connection_config.sqlite.connection_mode = 3\n",
    "print(\"ML metadata store is ready.\")\n",
    "\n",
    "beam_pipeline_args = [\n",
    "  f'--runner=DirectRunner',\n",
    "  f'--project={config.PROJECT_ID}',\n",
    "  f'--temp_location={config.ARTIFACT_STORE_URI}/beam/tmp'\n",
    "]\n",
    "\n",
    "pipeline_module.SCHEMA_DIR = 'tfx_pipeline/schema'\n",
    "pipeline_module.LOOKUP_EXPORTER_MODULE = 'tfx_pipeline/lookup_creator.py'\n",
    "pipeline_module.SCANN_INDEXER_MODULE = 'tfx_pipeline/scann_indexer.py'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "runner = BeamDagRunner()\n",
    "\n",
    "pipeline = pipeline_module.create_pipeline(\n",
    "  pipeline_name=config.PIPELINE_NAME,\n",
    "  pipeline_root=pipeline_root,\n",
    "  project_id=config.PROJECT_ID,\n",
    "  bq_dataset_name=config.BQ_DATASET_NAME,\n",
    "  min_item_frequency=15,\n",
    "  max_group_size=10,\n",
    "  dimensions=50,\n",
    "  num_leaves=500,\n",
    "  eval_min_recall=0.8,\n",
    "  eval_max_latency=0.001,\n",
    "  ai_platform_training_args=None,\n",
    "  beam_pipeline_args=beam_pipeline_args,\n",
    "  model_regisrty_uri=model_regisrty_uri,\n",
    "  metadata_connection_config=metadata_connection_config,\n",
    "  enable_cache=True\n",
    ")\n",<commit_msg>Rename lookup exporter to lookup creator<commit_after>{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TFX - Compile and Deploy the TFX Pipeline to KFP\n",
    "\n",
    "This Notebook helps you to compile the **TFX Pipeline** to a **KFP package**. This will creat an **Argo YAML** file in a **.tar.gz** package. We perform the following steps:\n",
    "1. Build a custom container image that include our modules\n",
    "2. Compile TFX Pipeline using CLI\n",
    "3. Deploy the compiled pipeline to KFP \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install -q -U kfp"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Set compile time variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "os.environ['PROJECT_ID'] = 'ksalama-cloudml' # Set your project.\n",
    "os.environ['BUCKET'] = 'ksalama-cloudml' # Set your bucket.\n",
    "os.environ['REGION'] = 'us-central1' # Set your region.\n",
    "os.environ['GKE_CLUSTER_NAME'] = 'ksalama-cloudml-dev' # Set your GKE cluster name.\n",
    "os.environ['GKE_CLUSTER_ZONE'] = 'europe-west2-a' # Set your GKE cluster zone.\n",
    "\n",
    "os.environ['IMAGE_NAME'] = 'tfx-ml'\n",
    "os.environ['TAG'] = 'tfx0.25.0'\n",
    "os.environ['ML_IMAGE_URI']=f'gcr.io/{os.environ.get(\"PROJECT_ID\")}/{os.environ.get(\"IMAGE_NAME\")}:{os.environ.get(\"TAG\")}'\n",
    "\n",
    "os.environ['NAMESPACE'] = 'kubeflow-pipelines'\n",
    "os.environ['ARTIFACT_STORE_URI'] = f'gs://{os.environ.get(\"BUCKET\")}/tfx_artifact_store'\n",
    "os.environ['GCS_STAGING_PATH'] = f'{os.environ.get(\"ARTIFACT_STORE_URI\")}/staging'\n",
    "\n",
    "os.environ['RUNTIME_VERSION'] = '2.2'\n",
    "os.environ['PYTHON_VERSION'] = '3.7'\n",
    "os.environ['BEAM_RUNNER'] = 'DirectRunner'\n",
    "os.environ['MODEL_REGISTRY_URI'] = f'{os.environ.get(\"ARTIFACT_STORE_URI\")}/model_registry'\n",
    "\n",
    "os.environ['PIPELINE_NAME'] = 'tfx_bqml_scann'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tfx_pipeline import config\n",
    "\n",
    "for key, value in config.__dict__.items():\n",
    "  if key.isupper(): print(f'{key}: {value}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run the Pipeline Locally using Beam Runner"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import kfp\n",
    "import tfx\n",
    "from tfx.orchestration.beam.beam_dag_runner import BeamDagRunner\n",
    "from tfx_pipeline import pipeline as pipeline_module\n",
    "import tensorflow as tf\n",
    "import ml_metadata as mlmd\n",
    "from ml_metadata.proto import metadata_store_pb2\n",
    "import logging\n",
    "\n",
    "logging.getLogger().setLevel(logging.INFO)\n",
    "\n",
    "print(\"TFX Version:\", tfx.__version__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_root = f'{config.ARTIFACT_STORE_URI}/{config.PIPELINE_NAME}_beamrunner'\n",
    "model_regisrty_uri = f'{config.MODEL_REGISTRY_URI}_beamrunner'\n",
    "local_mlmd_sqllite = 'mlmd/mlmd.sqllite'\n",
    "\n",
    "print(f'Pipeline artifacts root: {pipeline_root}')    \n",
    "print(f'Model registry location: {model_regisrty_uri}')  \n",
    "\n",
    "if tf.io.gfile.exists(pipeline_root):\n",
    "  print(\"Removing previous artifacts...\")\n",
    "  tf.io.gfile.rmtree(pipeline_root)\n",
    "if tf.io.gfile.exists('mlmd'):\n",
    "  print(\"Removing local mlmd SQLite...\")\n",
    "  tf.io.gfile.rmtree('mlmd')\n",
    "print(\"Creating mlmd directory...\")\n",
    "tf.io.gfile.mkdir('mlmd')\n",
    "\n",
    "metadata_connection_config = metadata_store_pb2.ConnectionConfig()\n",
    "metadata_connection_config.sqlite.filename_uri = local_mlmd_sqllite\n",
    "metadata_connection_config.sqlite.connection_mode = 3\n",
    "print(\"ML metadata store is ready.\")\n",
    "\n",
    "beam_pipeline_args = [\n",
    "  f'--runner=DirectRunner',\n",
    "  f'--project={config.PROJECT_ID}',\n",
    "  f'--temp_location={config.ARTIFACT_STORE_URI}/beam/tmp'\n",
    "]\n",
    "\n",
    "pipeline_module.SCHEMA_DIR = 'tfx_pipeline/schema'\n",
    "pipeline_module.LOOKUP_CREATOR_MODULE = 'tfx_pipeline/lookup_creator.py'\n",
    "pipeline_module.SCANN_INDEXER_MODULE = 'tfx_pipeline/scann_indexer.py'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "runner = BeamDagRunner()\n",
    "\n",
    "pipeline = pipeline_module.create_pipeline(\n",
    "  pipeline_name=config.PIPELINE_NAME,\n",
    "  pipeline_root=pipeline_root,\n",
    "  project_id=config.PROJECT_ID,\n",
    "  bq_dataset_name=config.BQ_DATASET_NAME,\n",
    "  min_item_frequency=15,\n",
    "  max_group_size=10,\n",
    "  dimensions=50,\n",
    "  num_leaves=500,\n",
    "  eval_min_recall=0.8,\n",
    "  eval_max_latency=0.001,\n",
    "  ai_platform_training_args=None,\n",
    "  beam_pipeline_args=beam_pipeline_args,\n",
    "  model_regisrty_uri=model_regisrty_uri,\n",
    "  metadata_connection_config=metadata_connection_config,\n",
    "  enable_cache=True\n",
    ")\n",
{'fraction_non_alphanumeric': 0.203487358326068, 'fraction_numerical': 0.005579773321708806, 'mean_word_length': 3.2643122676579925, 'pattern_counts': {'":': 96, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3226647', 'n_tokens_mistral': 4567, 'n_tokens_neox': 4568, 'n_words': 680}