Working with distributed workloads

apiVersion: kueue.x-k8s.io/v1beta1
kind: ResourceFlavor
metadata:
  name: default-flavor

$ oc apply -f default_flavor.yaml

apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "cluster-queue"
spec:
  namespaceSelector: {}  # match all.
  resourceGroups:
  - coveredResources: ["cpu", "memory", "nvidia.com/gpu"]
    flavors:
    - name: "default-flavor"
      resources:
      - name: "cpu"
        nominalQuota: 9
      - name: "memory"
        nominalQuota: 36Gi
      - name: "nvidia.com/gpu"
        nominalQuota: 5

$ oc apply -f cluster_queue.yaml

apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  namespace: test
  name: local-queue-test
  annotations:
    kueue.x-k8s.io/default-queue: 'true'
spec:
  clusterQueue: cluster-queue

$ oc apply -f local_queue.yaml

from codeflare_sdk import generate_cert

generate_cert.generate_tls_cert(cluster.config.name, cluster.config.namespace)
generate_cert.export_env(cluster.config.name, cluster.config.namespace)

local_queue="your_local_queue_name"

labels = {"exampleLabel1": "exampleLabel1Value", "exampleLabel2": "exampleLabel2Value"}

$ pip install kfp

from kfp import dsl


@dsl.component(
    base_image="registry.redhat.io/ubi8/python-39:latest",
    packages_to_install=['codeflare-sdk']
)


def ray_fn():
   import ray (1)
   from codeflare_sdk import Cluster, ClusterConfiguration, generate_cert (2)


   cluster = Cluster( (3)
       ClusterConfiguration(
           namespace="my_project", (4)
           name="raytest",
           num_workers=1,
           head_cpus="500m",
           min_memory=1,
           max_memory=1,
           worker_extended_resource_requests={“nvidia.com/gpu”: 1}, (5)
           image="quay.io/rhoai/ray:2.23.0-py39-cu121", (6)
           local_queue="local_queue_name", (7)
       )
   )


   print(cluster.status())
   cluster.up() (8)
   cluster.wait_ready() (9)
   print(cluster.status())
   print(cluster.details())


   ray_dashboard_uri = cluster.cluster_dashboard_uri()
   ray_cluster_uri = cluster.cluster_uri()
   print(ray_dashboard_uri, ray_cluster_uri)

   # Enable Ray client to connect to secure Ray cluster that has mTLS enabled
   generate_cert.generate_tls_cert(cluster.config.name, cluster.config.namespace) (10)
   generate_cert.export_env(cluster.config.name, cluster.config.namespace)


   ray.init(address=ray_cluster_uri)
   print("Ray cluster is up and running: ", ray.is_initialized())


   @ray.remote
   def train_fn(): (11)
       # complex training function
       return 100


   result = ray.get(train_fn.remote())
   assert 100 == result
   ray.shutdown()
   cluster.down() (12)
   auth.logout()
   return result


@dsl.pipeline( (13)
   name="Ray Simple Example",
   description="Ray Simple Example",
)


def ray_integration():
   ray_fn()


if __name__ == '__main__': (14)
    from kfp.compiler import Compiler
    Compiler().compile(ray_integration, 'compiled-example.yaml')

$ python compile_example.py

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: pytorchjob-role
rules:
  - apiGroups:
      - "kubeflow.org"
    resources:
      - pytorchjobs
    verbs:
      - create
      - delete
      - get
      - list
      - patch
      - update
      - watch
  - apiGroups:
      - "kubeflow.org"
    resources:
      - pytorchjobs/status
    verbs:
      - get

$ oc apply -f pytorchjob_role.yaml

apiVersion: v1
kind: ServiceAccount
metadata:
  name: pytorchjob-sa
  namespace: kfto

$ oc apply -f pytorchjob_serviceaccount.yaml

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: pytorchjob-rolebinding
subjects:
- kind: ServiceAccount
  name: pytorchjob-sa
  namespace: kfto
roleRef:
  kind: ClusterRole
  name: pytorchjob-role
  apiGroup: rbac.authorization.k8s.io

$ oc apply -f pytorchjob_rolebinding.yaml

kind: ConfigMap
apiVersion: v1
metadata:
  name: my-config-trainingjob
  namespace: kfto
data:
  config.json: |
    {
      "model_name_or_path": "bigscience/bloom-560m",
      "training_data_path": "/data/input/twitter_complaints.json",
      "output_dir": "/data/output/tuning/bloom-twitter",
      "num_train_epochs": 10.0,
      "per_device_train_batch_size": 4,
      "gradient_accumulation_steps": 4,
      "learning_rate": 1e-05,
      "response_template": "\n### Label:",
      "dataset_text_field": "output",
      "use_flash_attn": false
    }

$ oc apply -f config_trainingjob.yaml

kind: ConfigMap
apiVersion: v1
metadata:
  name: twitter-complaints
  namespace: kfto
data:
  twitter_complaints.json: |
    [
        {"Tweet text":"@HMRCcustomers No this is my first job","ID":0,"Label":2,"text_label":"no complaint","output":"### Text: @HMRCcustomers No this is my first job\n\n### Label: no complaint"},
        {"Tweet text":"@KristaMariePark Thank you for your interest! If you decide to cancel, you can call Customer Care at 1-800-NYTIMES.","ID":1,"Label":2,"text_label":"no complaint","output":"### Text: @KristaMariePark Thank you for your interest! If you decide to cancel, you can call Customer Care at 1-800-NYTIMES.\n\n### Label: no complaint"},
        {"Tweet text":"@EE On Rosneath Arial having good upload and download speeds but terrible latency 200ms. Why is this.","ID":3,"Label":1,"text_label":"complaint","output":"### Text: @EE On Rosneath Arial having good upload and download speeds but terrible latency 200ms. Why is this.\n\n### Label: complaint"},
        {"Tweet text":"Couples wallpaper, so cute. :) #BrothersAtHome","ID":4,"Label":2,"text_label":"no complaint","output":"### Text: Couples wallpaper, so cute. :) #BrothersAtHome\n\n### Label: no complaint"},
        {"Tweet text":"@mckelldogs This might just be me, but-- eyedrops? Artificial tears are so useful when you're sleep-deprived and sp… https:\/\/t.co\/WRtNsokblG","ID":5,"Label":2,"text_label":"no complaint","output":"### Text: @mckelldogs This might just be me, but-- eyedrops? Artificial tears are so useful when you're sleep-deprived and sp… https:\/\/t.co\/WRtNsokblG\n\n### Label: no complaint"},
        {"Tweet text":"@Yelp can we get the exact calculations for a business rating (for example if its 4 stars but actually 4.2) or do we use a 3rd party site?","ID":6,"Label":2,"text_label":"no complaint","output":"### Text: @Yelp can we get the exact calculations for a business rating (for example if its 4 stars but actually 4.2) or do we use a 3rd party site?\n\n### Label: no complaint"},
        {"Tweet text":"@nationalgridus I have no water and the bill is current and paid. Can you do something about this?","ID":7,"Label":1,"text_label":"complaint","output":"### Text: @nationalgridus I have no water and the bill is current and paid. Can you do something about this?\n\n### Label: complaint"},
        {"Tweet text":"@JenniferTilly Merry Christmas to as well. You get more stunning every year ��","ID":9,"Label":2,"text_label":"no complaint","output":"### Text: @JenniferTilly Merry Christmas to as well. You get more stunning every year ��\n\n### Label: no complaint"}
    ]

$ oc apply -f training_data.yaml

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: trained-model
  namespace: kfto
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 10Gi

$ oc apply -f trainedmodelpvc.yaml

apiVersion: kubeflow.org/v1
kind: PyTorchJob
metadata:
  name: kfto-demo
  namespace: kfto
spec:
  pytorchReplicaSpecs:
    Master:
      replicas: 1
      restartPolicy: Never
      template:
        spec:
          containers:
            - env:
                - name: SFT_TRAINER_CONFIG_JSON_PATH
                  value: /etc/config/config.json
              image: 'quay.io/modh/fms-hf-tuning:release'
              imagePullPolicy: IfNotPresent
              name: pytorch
              volumeMounts:
                - mountPath: /etc/config
                  name: config-volume
                - mountPath: /data/input
                  name: dataset-volume
                - mountPath: /data/output
                  name: model-volume
          volumes:
            - configMap:
                items:
                  - key: config.json
                    path: config.json
                name: my-config-trainingjob
              name: config-volume
            - configMap:
                name: twitter-complaints
              name: dataset-volume
            - name: model-volume
              persistentVolumeClaim:
                claimName: trained-model
  runPolicy:
    suspend: false

$ oc apply -f pytorchjob.yaml

status:
 conditions:
   - lastTransitionTime: '2024-05-29T13:05:09Z'
     message: 'couldn''t assign flavors to pod set small-group-jobtest12: insufficient quota for nvidia.com/gpu in flavor default-flavor in ClusterQueue'