Notebook 进阶开发

引用项目资源（.py 文件）

将通用的函数或类封装在独立的 .py 文件中，通过##@resource_reference{"自定义名.py"}引用MaxCompute资源的方式，实现代码的模块化与复用，提升代码的可维护性。

1. 创建并发布 Python 资源

   1. 在 DataWorks Data Studio左侧导航栏单击，进入资源管理。

   2. 在资源管理目录树下，右键单击目标目录或单击右上角的 + ，选择新建资源 > MaxCompute Python，命名为 my_utils.py。

   3. 在文件内容处，点击在线编辑，将您调试好的工具函数代码粘贴到代码编辑框并保存。

      # my_utils.py
      def greet(name):
          return f"Hello, {name} from resource file!"

   4. 单击工具栏中的保存，然后发布该资源，使其对开发和生产任务可见。

2. 在 Notebook 中引用资源

   在 Notebook 的Python 单元格的第一行，使用 ##@resource_reference 语法引用已发布的资源。

   ##@resource_reference{"my_utils.py"}
   # 如果资源在目录下，比如 my_folder/my_utils.py，也是通过##@resource_reference{"my_utils.py"}，无需带目录名称。
   from my_utils import greet
   
   message = greet('DataWorks')
   print(message)

3. 开发环境调试运行

   运行Python单元格，将打印如下结果：

   Hello, DataWorks from resource file!

   重要

   在开发环境调试运行时，如果识别到代码中包含##@resource_reference，系统将会把资源管理中的目标文件自动下载至个人目录的workspace/_dataworks/resource_references路径中，从而进行目标文件的引用。

   若报错ModuleNotFoundError，点击编辑器上方工具栏的重启按钮重新加载资源后再重试。

4. 发布至生产环境并验证

   保存并发布此 Notebook 节点后，前往运维中心 > 周期任务单击测试运行。任务成功后，您将在日志中看到输出 Hello, DataWorks from resource file!。

   重要

   若报错There is no file with id ...，请先将Python资源发布至生产环境。

更多操作，请参见MaxCompute资源与函数。

读写数据集（OSS/NAS）

在 Notebook 任务运行时，方便地读写存储在 OSS 或 NAS 上的大规模文件数据。

更多操作，请参见在个人开发环境中使用数据集。

引用工作空间参数

DataWorks 在已有调度参数的基础上引入工作空间参数，主要是为了解决跨任务、跨节点的全局配置复用与环境隔离问题。可在SQL单元格和Python单元格内以${workspace.param}格式来引用工作空间参数，其中param是您创建的工作空间参数名称。

1、创建工作空间参数：使用前，需前往 DataWorks 运维中心 > 调度设置 > 工作空间参数 页面创建所需参数。

2、引用工作空间参数：

• SQL单元格引用工作空间参数。

  SELECT '${workspace.param}';

  查询工作空间参数，运行成功后，将打印输出工作空间参数的具体赋值。

• Python单元格引用工作空间参数。

  print('${workspace.param}')

  输出工作空间参数，运行成功后，将打印输出工作空间参数的具体赋值。

更多详情，请参见使用工作空间参数。

连接 MaxCompute

• %odps：获取 PyODPS 入口对象

  此命令返回一个与当前MaxCompute项目绑定、已认证的 PyODPS 对象，避免在代码中硬编码 AccessKey，是与 MaxCompute 交互的推荐方式。

  1. 使用Magic Command创建MaxCompute连接。 输入%odps，右下角会出现MaxCompute的计算资源选择器入口（并自动选择计算资源）。点击右下角的MaxCompute项目名称，可更换MaxCompute项目。

     o=%odps 

  2. 使用获取到的MaxCompute计算资源运行PyODPS脚本。

     例如，获取当前项目下的所有表：

     with o.execute_sql('show tables').open_reader() as reader:
         print(reader.raw)

• %maxframe：建立 MaxFrame 连接

  此命令用于创建 MaxFrame 会话，为 MaxCompute 提供类似 Pandas 的分布式数据处理能力。

  # 连接并访问MaxCompute MaxFrame Session
  mf_session = %maxframe
  
  df = mf_session.read_odps_table('your_mc_table')
  print(df.head())
  
  # 开发调试结束后，应手动销毁会话以释放资源
  mf_session.destroy()

连接 Spark 计算资源

DataWorks Notebook 支持连接多种 Spark 引擎。不同引擎在连接方式、执行上下文和资源管理上存在差异。

连接 Lindorm Ray 计算资源

Lindorm计算引擎的RAY资源组提供分布式计算服务，支持AI负载端到端处理。通过 Magic Command，在 Notebook 中无缝连接 Lindorm Ray 资源，进行交互式开发、调试，并将其发布为生产调度任务。

1. 建立连接：在 Python 单元格中运行 %lindorm_ray 命令。运行后，单元格右下角将出现计算资源选择器，请在此处选择您的Lindorm 计算资源和创建好的Ray 资源组。

   # 连接到指定的Lindorm Ray资源组
   %lindorm_ray

   重要

   • 连接Lindorm Ray计算资源后，同一个 Notebook将不再支持运行SQL单元格。Lindorm Ray 引擎专注于执行 Python 和 Ray 代码。

   • 当您多次运行同一个代码单元格时，系统会自动终止上一个正在运行的 Ray Job，并启动一个新的 Job。这可以有效避免资源浪费和任务冲突。

2. 执行 Ray 代码：连接成功后，在新的 Python 单元格中直接编写和执行 Ray 代码。日志会实时回传到单元格的输出区域，方便您进行交互式调试。

   以下示例定义了一个简单的远程任务（使用 @ray.remote 装饰器），它会在 Ray 集群上执行，并将日志和最终结果返回到单元格的输出区。

   import ray
   import time
   
   @ray.remote
   def hello_world():
     print("Hello from Lindorm Ray!")
     time.sleep(5)
     return "Task finished."
   # 提交远程任务
   result_ref = hello_world.remote()
   print(ray.get(result_ref))

3. （可选）自定义启动参数：如需为 Ray 环境指定额外的配置，例如安装第三方 Python 包或上传本地代码文件，可以使用 %%lindorm_ray 命令建立连接。

   • 示例1：安装依赖包

     通过 pip 参数在 Ray 环境中安装 jieba 包。

     %%lindorm_ray
     {
       "runtime_env": {
         "pip": ["jieba"]
       }
     }

     环境准备就绪后，在后续的 Ray 任务中导入并使用该包。以下示例演示如何在远程函数中调用jieba进行中文分词：

     import ray 
     
     @ray.remote
     def do_work(x):
         import jieba
     
         return "/".join(jieba.cut(x))
     
     print(ray.get(do_work.remote("欢迎使用DataWorks+LindormRay解决方案")))

   • 示例2：上传并使用DataWorks资源

     通过 working_dir 参数，将 DataWorks 资源管理中已上传的资源进一步上传到 Ray 集群，使其可以在任务中被导入和调用。

     重要

     • 使用 working_dir 上传资源时，文件会直接从您的开发环境上传到 Ray 集群，存在 100MB 的大小限制。如果资源包过大，可能导致上传失败或 Ray 节点不稳定。

     • 对于体积较大的资源文件或依赖（超过100MB），建议您先将其上传到 OSS，然后在代码中从 OSS 下载和读取，或者直接将其打包到自定义镜像中。这能提供更好的稳定性和性能。

     # 引用Data Studio资源管理中上传的资源，声明路径
     %%lindorm_ray
     {
         "runtime_env": {
             "working_dir": "/mnt/workspace/_dataworks/resource_references"
         }
     }

     假设在DataWorks的Data Stuido资源管理中上传了一个ray_resource.py文件，编写并执行以下单元格时，系统会自动解析后续代码中的##@resource_reference 声明，并将下载相应资源到 /mnt/workspace/_dataworks/resource_references路径。

     ray_resource.py的代码内容示例如下：

     def fun():
         print("This is a test function in ray_resource.py")

     重要

     在开发环境中，当执行完包含##@resource_reference的单元格后，您需要重新运行上面的%%lindorm_ray单元格，会将下载好的资源包含在working_dir中上传到Ray集群。在生产环境中，您无需重新运行。

     import ray 
     
     ##@resource_reference{"ray_resource.py"}
     
     @ray.remote
     def do_work(x):
         print('Ray says:', x)
     
         from ray_resource import fun
         fun()
         return x
     
     worker = do_work.remote("欢迎使用DataWorks+LindormRay解决方案")
     print(ray.get(worker))

4. 生产调度与运维：开发调试完成后，您可以将此 Notebook 节点提交发布，它将作为一个 Lindorm Ray 节点在 DAG 中进行周期性调度。

   • 参数化：您的代码可以正常使用 DataWorks 的标准调度参数，例如 ${bizdate}。

   • 日志查看：在生产环境中，为了避免日志过多影响性能，系统默认只加载前 1MB 的日志。如果日志被截断，输出结果中会提供一个链接，引导您跳转至 Lindorm 控制台查看完整的任务日志。

   • 资源释放：生产调度任务结束后，Lindorm Ray任务的状态会变为终态，不再占用资源；交互式开发时，你可以通过重启Kernel、关闭Notebook将Lindorm Ray的任务变为终态。

附录：Magic Command 速查表