KFPv2 support step 1

.github/workflows/test.yml

@@ -62,7 +62,7 @@ jobs:
         steps:
             - name: Checkout
               uses: actions/checkout@v4
-            - name: Test KFP lib
+            - name: Test KFP v1 lib
               run: |
                   source kind/requirements.env
                   export PATH=$PATH:/tmp/
@@ -93,7 +93,7 @@ jobs:
                   sudo rm -rf /usr/share/dotnet /opt/ghc /usr/local/lib/android /usr/local/share/powershell /usr/share/swift /usr/lib/jvm /usr/local/.ghcup
                   sudo docker rmi $(docker image ls -aq) >/dev/null 2>&1 || true
                   df -h
-            - name: Test KFP worflow run
+            - name: Test KFP v1 worflow run
               timeout-minutes: 120
               run: |
                   source kind/requirements.env
@@ -108,6 +108,8 @@ jobs:
                   chmod 777 /tmp/kubectl
                   curl https://dl.min.io/client/mc/release/linux-amd64/mc --create-dirs -o /tmp/mc
                   chmod +x /tmp/mc
+                  export DEPLOY_KUBEFLOW=1
                   make -C kind setup
-                  make -C transforms workflow-build
+                  make -C kfp/kfp_support_lib test
+                  make -C transforms/universal/noop/ workflow-build
                   make -C transforms/universal/noop workflow-test

.make.defaults

@@ -53,6 +53,9 @@ KIND_CLUSTER_NAME=dataprep
 DPK_PYTHON_LIB_DIR=$(REPOROOT)/data-processing-lib/python
 DPK_RAY_LIB_DIR=$(REPOROOT)/data-processing-lib/ray
 DPK_SPARK_LIB_DIR=$(REPOROOT)/data-processing-lib/spark
+
+KFPv2?=0
+
 #######################################################################################
 # Lists all targets and optional help text found in the target.
 # Adapted from https://stackoverflow.com/a/65243296/45375
@@ -200,7 +203,7 @@ __check_defined = \
 	cp -p -R ${LIB_PATH}/README.md ${LIB_NAME}
 
 # Build and image using the local Dockerfile and make the data-processing-lib/python
-# available in the current directory for use by the Dockerfile (i.e. to install the library). 
+# available in the current directory for use by the Dockerfile (i.e. to install the library).
 .PHONY: .defaults.python-lib-src-image
 .defaults.python-lib-src-image:: # Must be called with a DOCKER_LOCAL_IMAGE= settings.
 	@# Help: Build the Python $(DOCKER_LOCAL_IMAGE) using the $(DOCKER_FILE), requirements.txt and data-processing-lib/python source
@@ -261,8 +264,8 @@ __check_defined = \
 
 # Install all source from the repo for a python runtime transform into an existing venv 
 .PHONY: .defaults.install-python-lib-src-venv
-.defaults.install-python-lib-src-venv:: 
-	@# Help: Install Python data processing library source into existing venv 
+.defaults.install-python-lib-src-venv::
+	@# Help: Install Python data processing library source into existing venv
 	@echo Installing Python data processing library source to existing venv
 	@source venv/bin/activate;				      	\
 	pip install pytest;						\
@@ -276,8 +279,8 @@ __check_defined = \
 
 # Install all source from the repo for a ray runtime transform into an existing venv 
 .PHONY: .defaults.install-ray-lib-src-venv
-.defaults.install-ray-lib-src-venv:: 
-	@# Help: Install Ray and Python data processing library source into existing venv 
+.defaults.install-ray-lib-src-venv::
+	@# Help: Install Ray and Python data processing library source into existing venv
 	@echo Installing Ray and Python data processing library source to existing venv
 	@source venv/bin/activate;				      	\
 	pip install pytest;						\
@@ -291,11 +294,10 @@ __check_defined = \
 .PHONY: .defaults.spark-lib-src-venv
 .defaults.spark-lib-src-venv:: .defaults.create-venv .defaults.install-spark-lib-src-venv .defaults.install-local-requirements-venv
 
-# Install all source from the repo for a spark runtime transform into an existing venv 
+# Install the python-based lib BEFORE spark assuming spark depends on the same version as python source.
 .PHONY: .defaults.install-spark-lib-src-venv
-.defaults.install-spark-lib-src-venv:: 
-	@# Help: Install Spark and Python data processing library source into existing venv 
-	@echo ""
+.defaults.install-spark-lib-src-venv::
+	@# Help: Install Spark and Python data processing library source into existing venv
 	@echo Installing Spark and Python data processing library source to existing venv
 	@source venv/bin/activate;				      	\
 	pip install pytest;						\

.make.versions

@@ -13,6 +13,7 @@ RELEASE_VERSION_SUFFIX=.dev6
 # Data prep lab wheel version
 DPK_LIB_VERSION=0.2.0$(RELEASE_VERSION_SUFFIX)
 DPK_LIB_KFP_VERSION=0.2.0$(RELEASE_VERSION_SUFFIX)
+DPK_LIB_KFP_VERSION_v2=0.2.0$(RELEASE_VERSION_SUFFIX)
 
 # Begin transform versions/tags
 BLOCKLIST_VERSION=0.4.2$(RELEASE_VERSION_SUFFIX)

kfp/doc/setup.md

@@ -66,7 +66,14 @@ choose your OS system, and process according to "(Optional) Install the MinIO Cl
 
 ## Installation steps <a name = "installation"></a>
 
-You can create a Kind cluster with all required software installed using the following command: 
+Before installation, you have to deside which KFP version do you want to use. 
+In order to use KFP v2, please set the following environment variable:
+
+```shell
+export KFPv2=1
+```
+
+Now, you can create a Kind cluster with all required software installed using the following command: 
 
 ```shell
  make setup

kfp/doc/simple_transform_pipeline.md

@@ -34,16 +34,16 @@ Note: the project and the explanation below are based on [KFPv1](https://www.kub
 * Pipeline wiring - definition of the sequence of invocation (with parameter passing) of participating components
 * Additional configuration
 
-### Imports definition <a name = "imports"></a> 
+### Imports definition <a name = "imports"></a>
 
 ```python
 import kfp.compiler as compiler
 import kfp.components as comp
 import kfp.dsl as dsl
-from kfp_support.workflow_support.utils import (
-    ONE_HOUR_SEC,
-    ONE_WEEK_SEC,
-    ComponentUtils,
+from kfp_support.workflow_support.runtime_utils import (
+  ONE_HOUR_SEC,
+  ONE_WEEK_SEC,
+  ComponentUtils,
 )
 from kubernetes import client as k8s_client
 ```
@@ -73,8 +73,8 @@ Ray cluster. For each step we have to define a component that will execute them:
 Note: here we are using shared components described in this [document](../kfp_ray_components/README.md) for `create_ray_op`, 
 `execute_ray_jobs_op` and `cleanup_ray_op`,  while `compute_exec_params_op` component is built inline, because it might
 differ significantly. For "simple" pipeline cases we can use the 
-[default implementation](../kfp_support_lib/src/kfp_support/workflow_support/utils/remote_jobs_utils.py),
-while, for example for exact dedup, we are using a very [specialized one](../transform_workflows/universal/ededup/src/ededup_compute_execution_params.py).
+[default implementation](../kfp_support_lib/src/kfp_support/workflow_support/runtime_utils/remote_jobs_utils.py),
+while, for example for exact dedup, we are using a very [specialized one](../../transforms/universal/ededup/kfp_ray/v2/src/ededup_compute_execution_params.py).
 
 ### Input parameters definition <a name = "inputs"></a> 
 

kfp/kfp_ray_components/Dockerfile

@@ -10,15 +10,19 @@ LABEL git-commit=$GIT_COMMIT
 COPY requirements.txt requirements.txt
 RUN pip install --no-cache-dir -r  requirements.txt
 
-# Copy and install data processing libraries 
+# Copy and install data processing libraries
 # These are expected to be placed in the docker context before this is run (see the make image).
 COPY --chown=ray:users data-processing-lib-python/ data-processing-lib-python/
 RUN cd data-processing-lib-python && pip install --no-cache-dir -e .
-COPY --chown=ray:users data-processing-lib-ray/ data-processing-lib-ray/ 
+COPY --chown=ray:users data-processing-lib-ray/ data-processing-lib-ray/
 RUN cd data-processing-lib-ray    && pip install --no-cache-dir -e .
 
-COPY --chown=ray:users kfp_support_lib/ kfp_support_lib/
-RUN cd kfp_support_lib && pip install --no-cache-dir -e .
+COPY --chown=ray:users python_apiserver_client python_apiserver_client/
+RUN cd python_apiserver_client && pip install --no-cache-dir -e .
+
+COPY --chown=ray:users workflow_support_lib workflow_support_lib/
+RUN cd workflow_support_lib && pip install --no-cache-dir -e .
+ENV KFP_v2=$KFP_v2
 # remove credentials-containing file
 RUN rm requirements.txt
 # components

kfp/kfp_ray_components/Makefile

@@ -2,26 +2,39 @@
 # # know where they are running from.
 REPOROOT=../..
 
+# Include the common rules.
+# Use "make help" to see them.
+include $(REPOROOT)/.make.defaults
+
 IGNORE := $(shell bash -c "sed -n /=/p  ${REPOROOT}/kfp/requirements.env | sed 's/=/:=/' | sed 's/^/export /' > makeenv")
 
 include makeenv
-DOCKER_FILE=Dockerfile
+
+ifeq ($(KFPv2), 1)
+DOCKER_IMAGE_NAME=kfp-data-processing_v2
+DOCKER_IMAGE_VERSION=${KFP_DOCKER_VERSION_v2}
+WORKFLOW_SUPPORT_LIB=kfp_v2_workflow_support
+else
 DOCKER_IMAGE_NAME=kfp-data-processing
 DOCKER_IMAGE_VERSION=${KFP_DOCKER_VERSION}
+WORKFLOW_SUPPORT_LIB=kfp_v1_workflow_support
+endif
 
-# Include the common rules.
-# Use "make help" to see them.
-include $(REPOROOT)/.make.defaults
+
+#DOCKER_IMG=${DOCKER_HOSTNAME}/${DOCKER_NAMESPACE}/${DOCKER_IMAGE_NAME}:${DOCKER_IMAGE_VERSION}
+DOCKER_IMG=$(DOCKER_LOCAL_IMAGE)
 
 .PHONY: .lib-src-image
 .lib-src-image:: 
 	$(MAKE) .defaults.copy-lib LIB_PATH=$(DPK_RAY_LIB_DIR) LIB_NAME=data-processing-lib-ray
 	$(MAKE) .defaults.copy-lib LIB_PATH=$(DPK_PYTHON_LIB_DIR) LIB_NAME=data-processing-lib-python
-	$(MAKE) .defaults.copy-lib LIB_PATH=$(REPOROOT)/kfp/kfp_support_lib LIB_NAME=kfp_support_lib
+	$(MAKE) .defaults.copy-lib LIB_PATH=$(REPOROOT)/kfp/kfp_support_lib/python_apiserver_client LIB_NAME=python_apiserver_client
+	$(MAKE) .defaults.copy-lib LIB_PATH=$(REPOROOT)/kfp/kfp_support_lib/$(WORKFLOW_SUPPORT_LIB) LIB_NAME=workflow_support_lib
 	$(MAKE) .defaults.image
 	rm -rf data-processing-lib-ray
 	rm -rf data-processing-lib-python
-	rm -rf kfp_support_lib
+	rm -rf python_apiserver_client
+	rm -rf workflow_support_lib
 
 .PHONY: image
 image: Dockerfile requirements.txt
@@ -34,11 +47,12 @@ set-versions::  reconcile-requirements
 .PHONY: reconcile-requirements
 reconcile-requirements:
 	@# Help: Update yaml files to build images tagged as version $(KFP_DOCKER_VERSION)
-	sed -i.back "s/kfp-data-processing:[0-9].*/kfp-data-processing:${KFP_DOCKER_VERSION}/" createRayClusterComponent.yaml
-	sed -i.back "s/kfp-data-processing:[0-9].*/kfp-data-processing:${KFP_DOCKER_VERSION}/" deleteRayClusterComponent.yaml
-	sed -i.back "s/kfp-data-processing:[0-9].*/kfp-data-processing:${KFP_DOCKER_VERSION}/" executeRayJobComponent.yaml
-	sed -i.back "s/kfp-data-processing:[0-9].*/kfp-data-processing:${KFP_DOCKER_VERSION}/" executeRayJobComponent_multi_s3.yaml
-	sed -i.back "s/kfp-data-processing:[0-9].*/kfp-data-processing:${KFP_DOCKER_VERSION}/" executeSubWorkflowComponent.yaml
+	sed -i.back "s/kfp-data-processing.*:[0-9].*/$(DOCKER_IMAGE_NAME):${KFP_DOCKER_VERSION}/" createRayClusterComponent.yaml
+	sed -i.back "s/kfp-data-processing.*:[0-9].*/$(DOCKER_IMAGE_NAME):${KFP_DOCKER_VERSION}/" deleteRayClusterComponent.yaml
+	sed -i.back "s/kfp-data-processing.*:[0-9].*/$(DOCKER_IMAGE_NAME):${KFP_DOCKER_VERSION}/" executeRayJobComponent.yaml
+	sed -i.back "s/kfp-data-processing.*:[0-9].*/$(DOCKER_IMAGE_NAME):${KFP_DOCKER_VERSION}/" executeRayJobComponent_multi_s3.yaml
+	# TODO remove it for KFPv2
+	sed -i.back "s/kfp-data-processing*:[0-9].*/$(DOCKER_IMAGE_NAME):${KFP_DOCKER_VERSION}/" executeSubWorkflowComponent.yaml
 
 .PHONY: load-image
 load-image:

kfp/kfp_ray_components/executeRayJobComponent.yaml

@@ -6,7 +6,7 @@ inputs:
     - { name: run_id, type: String, description: "The KFP Run ID" }
     - { name: additional_params, type: String, description: "additional parameters" }
     # The component converts the dictionary to json string
-    - { name: exec_params, type: dict, description: "job parameters" }
+    - { name: exec_params, type: JsonObject, description: "job parameters" }
     - { name: exec_script_name, type: String, description: "transform script name" }
     - { name: server_url, type: String, default: "", description: "url of api server" }
 

kfp/kfp_ray_components/executeRayJobComponent_multi_s3.yaml

@@ -8,7 +8,7 @@ inputs:
     - { name: server_url, type: String, default: "", description: "url of api server" }
     - { name: prefix, type: String, default: "", description: "prefix for extra credentials" }
     # The component converts the dictionary to json string
-    - { name: exec_params, type: dict, description: "job parameters" }
+    - { name: exec_params, type: JsonObject, description: "job parameters" }
     - { name: additional_params, type: String, description: "additional parameters" }
 
 implementation:

kfp/kfp_ray_components/src/create_ray_cluster.py

@@ -9,10 +9,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 ################################################################################
-
 import sys
-
-from kfp_support.workflow_support.utils import KFPUtils, RayRemoteJobs
+from workflow_support.runtime_utils import KFPUtils,  RayRemoteJobs
 
 
 def start_ray_cluster(

kfp/kfp_ray_components/src/delete_ray_cluster.py

@@ -9,11 +9,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 ################################################################################
-
 import sys
-
-from kfp_support.workflow_support.utils import KFPUtils, RayRemoteJobs
-
+from workflow_support.runtime_utils import KFPUtils, RayRemoteJobs
 
 # Cleans and shutdowns the Ray cluster
 def cleanup_ray_cluster(

kfp/kfp_ray_components/src/execute_ray_job.py

@@ -9,9 +9,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 ################################################################################
-
-from kfp_support.workflow_support.utils import KFPUtils, execute_ray_jobs
-
+from workflow_support.runtime_utils import KFPUtils, execute_ray_jobs
 
 if __name__ == "__main__":
     import argparse

kfp/kfp_ray_components/src/execute_ray_job_multi_s3.py

@@ -10,8 +10,7 @@
 # limitations under the License.
 ################################################################################
 
-from kfp_support.workflow_support.utils import KFPUtils, execute_ray_jobs
-
+from workflow_support.runtime_utils import KFPUtils, execute_ray_jobs
 
 if __name__ == "__main__":
     import argparse

kfp/kfp_ray_components/src/subworkflow.py

@@ -1,9 +1,11 @@
 import sys
 
-from data_processing.utils.params_utils import ParamsUtils
-from kfp_support.workflow_support.utils import KFPUtils, PipelinesUtils
+from workflow_support.runtime_utils import KFPUtils
+from workflow_support.pipeline_utils import PipelinesUtils
 
 
+from data_processing.utils import ParamsUtils
+
 def invoke_sub_workflow(
     name: str,  # workflow name
     prefix: str,  # workflow arguments prefix