Databricks 中的检查点¶

先决条件¶

在 Databricks 中使用 Snowpark Checkpoints 之前，请确保环境满足以下要求：

• `PySpark：`版本 3.5.0 或更高版本。

• `Python：`版本 3.9、3.10 和 3.11

满足这些要求的 Databricks Runtime 版本是：

• Databricks Runtime 14.3 LTS

• Databricks Runtime 15.4 LTS

输入/输出 (I/O) 策略¶

为确保 Snowpark Checkpoints 在各种环境中正常运行，可以使用接口 EnvStrategy 及其实现类进行文件读写操作。这使得 I/O 操作具有适应性和自定义性。

• 使用 Snowpark Checkpoints，您可以通过创建实现 EnvStrategy 接口的类来实现自己的自定义输入/输出方法。然后，您可以根据特定的执行环境和预期定制操作。

• 在内部，该包使用默认类 (IODefaultStrategy) 来实现 EnvStrategy 接口并提供 I/O 操作的基本实现。您可以将此默认策略替换为适合环境特定需求的自定义实现。

I/O 函数¶

这些文件读写方法可以自定义：

• mkdir：创建文件夹。

• folder_exists：检查文件夹是否存在。

• file_exists：检查文件是否存在。

• write：将内容写入到文件。

• read：从文件中读取内容。

• read_bytes：从文件中读取二进制内容。

• ls：列出目录的内容。

• getcwd：获取当前工作目录。

• remove_dir：移除目录及其内容。此功能仅在 snowpark-checkpoints-collectors 模块中使用。

• telemetry_path_files：获取遥测文件的路径。

Databricks 策略¶

Databricks 策略是用于了解如何使用 DBFS 文件路径的配置。它使用 normalize_dbfs_path 函数来确保所有路径都以 /dbfs/ 开头。

如何使用该策略¶

要使用 Databricks 策略，必须在代码中对其进行明确配置。方法如下：

1. 导入所需的类：

   from typing import Optional, BinaryIO
   from pathlib import Path
   from snowflake.snowpark_checkpoints_collector.io_utils import EnvStrategy, IODefaultStrategy
   from snowflake.snowpark_checkpoints_collector.io_utils.io_file_manager import get_io_file_manager
   

   Copy

2. 定义 Databricks 策略：

   class IODatabricksStrategy(EnvStrategy):
   
     def __init__(self):
         self.default_strategy = IODefaultStrategy()
   
     def mkdir(self, path: str, exist_ok: bool = False) -> None:
         path = normalize_dbfs_path(path)
         self.default_strategy.mkdir(path, exist_ok=exist_ok)
   
     def folder_exists(self, path: str) -> bool:
         path = normalize_dbfs_path(path)
         return self.default_strategy.folder_exists(path)
   
     def file_exists(self, path: str) -> bool:
         path = normalize_dbfs_path(path)
         return self.default_strategy.file_exists(path)
   
     def write(self, file_path: str, file_content: str, overwrite: bool = True) -> None:
         file_path = normalize_dbfs_path(file_path)
         self.default_strategy.write(file_path, file_content, overwrite=overwrite)
   
     def read(
         self, file_path: str, mode: str = "r", encoding: Optional[str] = None
     ) -> str:
         file_path = normalize_dbfs_path(file_path)
         return self.default_strategy.read(file_path, mode=mode, encoding=encoding)
   
     def read_bytes(self, file_path: str) -> bytes:
         file_path = normalize_dbfs_path(file_path)
         return self.default_strategy.read_bytes(file_path)
   
     def ls(self, path: str, recursive: bool = False) -> list[str]:
         file_path = normalize_dbfs_path(path)
         list_of_files = self.default_strategy.ls(file_path, recursive=recursive)
         return [content.replace("/dbfs","") for content in list_of_files]
   
     def getcwd(self) -> str:
         try:
             parent_folder = "/snowpark_checkpoints"
             self.mkdir(parent_folder, exist_ok=True)
             return parent_folder
         except Exception:
             return ""
   
     def remove_dir(self, path:str) -> None:
         path = normalize_dbfs_path(path)
         self.default_strategy.remove_dir(path)
   
     def telemetry_path_files(self, path:str) -> Path:
         path = normalize_dbfs_path(path)
         return self.default_strategy.telemetry_path_files(path)
   
   def normalize_dbfs_path(path: str) -> str:
       if isinstance(path, Path):
           path = str(path)
       if not path.startswith("/"):
           path = "/" + path
       if not path.startswith("/dbfs/"):
           path = f'/dbfs{path}'
       return path
   

   Copy

3. 配置 Databricks 策略：

   get_io_file_manager().set_strategy(IODatabricksStrategy())
   

   Copy

在 Databricks 脚本或笔记本开始时执行此代码会将 Snowpark Checkpoints 配置为使用定义的 I/O 策略在 DBFS 中正确处理文件。

可选自定义¶

对于更专业的输入/输出操作，可以设计和实现自定义策略。此方法提供了对 I/O 行为的完全控制和灵活性。此方法允许开发者根据其特定要求和约束来精确地定制策略，从而有可能优化性能、资源利用率或其他相关因素。