从 Git 文件夹或工作区文件导入 Python 模块

可以将 Python 代码存储在 Databricks Git 文件夹或工作区文件中,然后将该 Python 代码导入到 Delta Live Tables 管道中。 有关在 Databricks Git 文件夹或工作区文件中使用模块的详细信息,请参阅使用 Python 和 R 模块

注意

无法从存储在 Databricks Git 文件夹或工作区文件中的笔记本导入源代码。 可以在创建或编辑管道时直接添加笔记本。 请参阅创建管道

将 Python 模块导入到增量实时表管道

以下示例演示如何从工作区文件将数据集查询作为 Python 模块导入。 尽管此示例说明了如何使用工作区文件来存储管道源代码,但你可以将其与存储在 Git 文件夹中的源代码配合使用。

若要运行此示例,请使用以下步骤:

  1. 单击 Azure Databricks 工作区边栏中的 工作区图标 工作区以打开工作区浏览器。

  2. 使用工作区浏览器选择 Python 模块的目录。

  3. 单击所选目录最右侧列中的 Kebab 菜单,然后单击“创建”>“文件”

  4. 输入文件的名称,例如 clickstream_raw_module.py。 文件编辑器随即打开。 若要创建一个用于将源数据读入表中的模块,请在编辑器窗口中输入以下内容:

    from dlt import *
    
    json_path = "/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json"
    
    def create_clickstream_raw_table(spark):
      @table
      def clickstream_raw():
        return (
          spark.read.json(json_path)
        )
    
  5. 若要创建一个模块来创建包含已准备好的数据的新表,请在同一目录中创建一个新文件,输入该文件的名称(例如 clickstream_prepared_module.py),然后在新编辑器窗口中输入以下内容:

    from clickstream_raw_module import *
    from dlt import read
    from pyspark.sql.functions import *
    from pyspark.sql.types import *
    
    def create_clickstream_prepared_table(spark):
      create_clickstream_raw_table(spark)
      @table
      @expect("valid_current_page_title", "current_page_title IS NOT NULL")
      @expect_or_fail("valid_count", "click_count > 0")
      def clickstream_prepared():
        return (
          read("clickstream_raw")
            .withColumn("click_count", expr("CAST(n AS INT)"))
            .withColumnRenamed("curr_title", "current_page_title")
            .withColumnRenamed("prev_title", "previous_page_title")
            .select("current_page_title", "click_count", "previous_page_title")
        )
    
  6. 接下来,创建一个管道笔记本。 转到 Azure Databricks 登陆页,选择“创建笔记本”,或单击边栏中的新建图标“新建”,然后选择“笔记本”。 还可以在工作区浏览器中创建笔记本,方法是单击 Kebab 菜单,然后单击“创建”>“笔记本”

  7. 为笔记本命名,并确认“Python”是默认语言

  8. 单击 “创建”

  9. 在笔记本中输入示例代码。

    注意

    如果笔记本从工作区文件路径或 Git 文件夹路径导入与笔记本目录不同的模块或包,则必须使用 sys.path.append() 手动将路径追加到文件。

    如果从 Git 文件夹导入文件,则必须在路径前面添加 /Workspace/。 例如 sys.path.append('/Workspace/...')。 省略路径中的 /Workspace/ 会导致错误。

    如果模块或包存储在笔记本所在的同一目录中,则无需手动追加路径。 从 Git 文件夹的根目录导入时,也不需要手动追加路径,因为根目录会自动追加到路径。

    import sys, os
    # You can omit the sys.path.append() statement when the imports are from the same directory as the notebook.
    sys.path.append(os.path.abspath('<module-path>'))
    
    import dlt
    from clickstream_prepared_module import *
    from pyspark.sql.functions import *
    from pyspark.sql.types import *
    
    create_clickstream_prepared_table(spark)
    
    @dlt.table(
      comment="A table containing the top pages linking to the Apache Spark page."
    )
    def top_spark_referrers():
      return (
        dlt.read("clickstream_prepared")
          .filter(expr("current_page_title == 'Apache_Spark'"))
          .withColumnRenamed("previous_page_title", "referrer")
          .sort(desc("click_count"))
          .select("referrer", "click_count")
          .limit(10)
      )
    

    <module-path> 替换为包含要导入的 Python 模块的目录的路径。

  10. 使用新笔记本创建管道。

  11. 若要运行管道,请在“管道详细信息”页中单击“启动”。

还可以将 Python 代码作为包导入。 增量实时表笔记本中的以下代码片段从笔记本所在目录中的 dlt_packages 目录导入 test_utils 包。 dlt_packages 目录包含文件 test_utils.py__init__.pytest_utils.py 定义函数 create_test_table()

import dlt

@dlt.table
def my_table():
  return dlt.read(...)

# ...

import dlt_packages.test_utils as test_utils
test_utils.create_test_table(spark)