Как изменить CustomJob для определенного компонента в vertexAI

Question

У меня возникли проблемы с изменением одного компонента в моем конвейере. Более подробно, компонент, который представляет собой компонент многократного использования, загружаемый через файл .yaml, должен работать в другой сети (VPC), чем сеть, назначенная VertexAI по умолчанию. Я не знаю, как это сделать.

Это более или менее мой подход на данный момент:

      # loading component
component_item = kfp.components.load_component(filename)

Это возвращает фабричную функцию со строго типизированной сигнатурой. После вызова с необходимыми аргументами фабрика создает экземпляр задачи конвейера (ContainerOp) (source). Поскольку .yaml моего компонента определяет входные и выходные данные, теперь я могу вызвать фабричную функцию, чтобы сделать ее настоящим ContainerOP, чтобы ее можно было использовать в дальнейшем в моем конвейере VertexAI.

      component_item_op = component_item(
        some_argument1=1,
        some_argument2=2,
        some_argument3=3,
        test=True,
    )

Обратите внимание, что компонент_item _op содержит класс <class 'kfp.dsl._container_op.ContainerOp'>

Пока все отлично, но, бегло просмотрев документацию kubeflow и skd google-cloud-aiplatform, я не нашел способа изменить «задание» так, чтобы я мог назначить ему другую сеть. Я знаю, что могу изменить сеть, если у меня будет доступ к объекту customjob.

Я пробовал разные подходы, но ни один из них пока не работал. В одном примере я пытался использовать create_custom_training_job_op_from_comComponent как часть пакета компонентов облачных конвейеров Google. Однако передача моего компонента_item_op не работает и завершается ошибкой:

      test = create_custom_training_job_op_from_component(
    component_item(
        some_argument1=1,
        some_argument2=2,
        some_argument3=3,
        test=True,
    )
)

      training_job_op_from_component
    if component_spec.component_spec.inputs:
AttributeError: inputs

Я ожидал, что это будет работать «из коробки», так как согласно документации можно передать объект типа ContainerOp, а все остальное оставить по умолчанию.

Судя по ошибке, похоже, что ему не удалось получить входные данные. Так что да, я здесь немного озадачен. Может быть, кто-нибудь подскажет мне еще что-нибудь или поможет.

.yaml

      name: some name
description: some desc

inputs:
- {name: some_argument1, type: Integer, description: "some desc"}
- {name: some_argument2, type: Integer, description: "some desc"}
- {name: some_argument3, type: Integer, description: "some desc"}
- {name: test, type: Boolean, description: 'runs a test when true'}

outputs:
- {name: some_out, type: Dataset, description: 'some desc'}

implementation:
  container:
      args:
      - --executor_input
      - executorInput: null
      - --function_to_execute
      - some_func
      command:
      - python3
      - -m
      - kfp.v2.components.executor_main
      - --component_module_path
      - some_file.py
      image: some_image_path

Установленные версии

      kfp                              1.8.19
kfp-pipeline-spec                0.1.16
kfp-server-api                   1.8.5

Answer 1

Что ж, оказывается, я делал правильные вещи, хотя и не в том порядке. Вот как я это исправил:

      tmp_op = create_custom_training_job_op_from_component(component_item, network="your_vpc_network_path)
component_op = tmp_op(project="your_project", some_argument1=1, some_argument2=2, some_argument3=3)

На данный момент у меня есть объект ContainerOP с правильными спецификациями worker_pool_specs.