6.4. 从数据科学项目运行分布式数据科学工作负载

步骤

1. 创建数据连接以将对象存储连接到您的数据科学项目，如 向数据科学项目添加数据连接 中所述。
2. 将管道服务器配置为使用数据连接，如 配置管道服务器 中所述。

3. 按如下方式创建数据科学项目管道：

   1. 安装 kfp-tekton Python 软件包，这是所有管道都需要的：

      $ pip install kfp-tekton

      Copy to Clipboard Toggle word wrap
   2. 安装管道所需的任何其他依赖项。

   3. 在 Python 代码中构建您的数据科学项目管道。例如，使用以下内容创建一个名为 compile_example.py 的文件：

      from kfp import components, dsl
      
      
      def ray_fn(openshift_server: str, openshift_token: str) -> int: 

      1

      
         import ray
         from codeflare_sdk.cluster.auth import TokenAuthentication
         from codeflare_sdk.cluster.cluster import Cluster, ClusterConfiguration
      
      
         auth = TokenAuthentication( 

      2

      
             token=openshift_token, server=openshift_server, skip_tls=True
         )
         auth_return = auth.login()
         cluster = Cluster( 

      3

      
             ClusterConfiguration(
                 name="raytest",
                 # namespace must exist
                 namespace="pipeline-example",
                 num_workers=1,
                 head_cpus="500m",
                 min_memory=1,
                 max_memory=1,
                 num_gpus=0,
                 image="quay.io/project-codeflare/ray:latest-py39-cu118", 

      4

      
                 instascale=False, 

      5

      
             )
         )
      
      
         print(cluster.status())
         cluster.up() 

      6

      
         cluster.wait_ready() 

      7

      
         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)
      
      
         # Before proceeding, ensure that the cluster exists and that its URI contains a value
         assert ray_cluster_uri, "Ray cluster must be started and set before proceeding"
      
      
         ray.init(address=ray_cluster_uri)
         print("Ray cluster is up and running: ", ray.is_initialized())
      
      
         @ray.remote
         def train_fn(): 

      8

      
             # complex training function
             return 100
      
      
         result = ray.get(train_fn.remote())
         assert 100 == result
         ray.shutdown()
         cluster.down() 

      9

      
         auth.logout()
         return result
      
      
      @dsl.pipeline( 

      10

      
         name="Ray Simple Example",
         description="Ray Simple Example",
      )
      def ray_integration(openshift_server, openshift_token):
         ray_op = components.create_component_from_func(
             ray_fn,
             base_image='registry.redhat.io/ubi8/python-39:latest',
             packages_to_install=["codeflare-sdk"],
         )
         ray_op(openshift_server, openshift_token)
      
      
      if __name__ == '__main__': 

      11

      
          from kfp_tekton.compiler import TektonCompiler
          TektonCompiler().compile(ray_integration, 'compiled-example.yaml')

      Copy to Clipboard Toggle word wrap

      1

      从 CodeFlare SDK 导入定义集群功能的软件包

      2

      使用您在创建管道运行时指定的值通过集群进行身份验证

      3

      指定 Ray 集群资源：将这些示例值替换为您的 Ray 集群的值

      4

      指定 Ray 集群镜像的位置：如果使用断开连接的环境，请将默认值替换为您的环境的位置

      5

      此发行版本不支持 InstaScale

      6

      使用指定的镜像和配置创建 Ray 集群

      7

      等待 Ray 集群就绪，然后继续

      8

      将本节中的示例详情替换为您的工作负载详情

      9

      工作负载完成后删除 Ray 集群

      10

      将示例名称和描述替换为您的工作负载的值

      11

      编译 Python 代码，并将输出保存在 YAML 文件中

   4. 编译 Python 文件（本例中为 compile_example.py 文件）：

      $ python compile_example.py

      Copy to Clipboard Toggle word wrap

      这个命令会创建一个 YAML 文件（在这个示例中是 compiled-example.yaml），您可以在下一步中导入该文件。

4. 导入您的数据科学管道，如 导入数据科学管道 中所述。
5. 调度管道运行，如 调度管道运行 中所述。
6. 当管道运行完成后，确认它包含在触发的管道运行列表中，如 查看触发的管道运行 中所述。