ModuleNotFoundError：运行 Delta Live Tables Python 笔记本时出现无名为“dlt”的模块错误

Question

当尝试创建 Python 笔记本并按照设置 databricks delta live 表的各种示例进行操作时，如果您尝试运行笔记本，您将立即遇到以下错误：

ModuleNotFoundError：没有名为“dlt”的模块

自给自足的开发人员可能会尝试使用“神奇命令”来安装所述模块来解决此问题：

%pip install dlt

但是可惜，这个 dlt 包与 databricks 增量实时表没有任何关系。运行您的代码现在将引发错误：

AttributeError：模块“dlt”没有属性“table”

（或类似的错误，取决于您尝试使用的第一个 dlt 类成员）

发生什么事了？如何运行 Delta Live Tables 管道设置代码？

Answer 1

虽然您需要在 databricks 笔记本环境中编写增量实时表设置代码，但您

不打算在那里运行它。运行代码的唯一受支持的方法是前往 pipelines 接口 来运行它。

回答结束。

虽然....

---

这对于编写大量代码并且甚至不确定其语法是否有效的开发人员来说是个坏消息（因为 databricks IDE 仅具有有限的实时反馈）。现在，您将陷入等待管道启动资源、启动、失败，然后通过堆栈跟踪来尝试找出出错的地方。您必须遵循此工作流程来解决逻辑错误，但在解决语法错误时不必遵循此工作流程。

这是我想出的解决方法：

try:
  import dlt # When run in a pipeline, this package will exist (no way to import it here)
except ImportError:
  class dlt: # "Mock" the dlt class so that we can syntax check the rest of our python in the databricks notebook editor
    def table(comment, **options): # Mock the @dlt.table attribute so that it is seen as syntactically valid below
      def _(f):
        pass
      return _;

@dlt.table(comment = "Raw Widget Data")
def widgets_raw():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv").option("header", "true").option("sep", "|")
      .load("/mnt/LandingZone/EMRAW/widgets")
  )

这里的技巧是我将 dlt 类模拟到最低限度以通过语法检查，因此可以验证我的其余代码。

令人烦恼的是sql笔记本没有这个问题，当你运行它们时，你会得到令人愉快的消息：

此 Delta Live Tables 查询在语法上是有效的，但您必须创建一个管道才能定义和填充表。

不幸的是，我发现 sql 笔记本在其他方面有限制，所以选择你的毒药。

无论哪种方式，希望你的代码在管道中运行之前不会真正

做任何事情。该笔记本仅用于设置，在必须从管道 UI 开始故障排除之前，最好先进行尽可能多的语法检查。

Answer 2

try:
    import dlt  # type: ignore
except ImportError:
    from unittest.mock import MagicMock
    class MockDlt:
        def __getattr__(self, name):
            return MagicMock()

        def __call__(self, *args, **kwargs):
            return MagicMock()
    dlt = MockDlt()

@dlt.table(comment = "Raw Widget Data")
@dlt.whatever(...)  # Supports any functionality
def widgets_raw():
    return (
        spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "csv")
            .option("header", "true").option("sep", "|")
            .load("/mnt/LandingZone/EMRAW/widgets")
    )

Answer 3

import dlt

语句并将其添加到使用以下代码的 sys 模块中。

import sys
from unittest.mock import patch, Mock

with patch.dict('sys.modules', {'dlt': Mock()}):
    import <module_with_dlt>