python - 在 databricks 笔记本中导入 Python 文件

Question

我正在尝试从 databricks 笔记本导入 python 文件。我能够导入该函数，但它一直给出错误 - NameError: name 'col' is not Defined。

我有一个文件/Workspace/Shared/Common/common.py。该文件中的代码是:

def cast_datatypes(df_target, df_source):
    cols = df_target.columns
    types = [f.dataType for f in df_target.schema.fields]
    for i, c in enumerate(cols):
        df_source = df_source.withColumn(c, col(c).cast(types[i]))
    df_source = df_source.select(cols)
    return df_source

我从笔记本上调用这个函数，并执行了以下方法

方法一:

import sys
sys.path.append("/Workspace/Shared/Common/common.py")
from common import cast_datatypes
from pyspark.sql.functions import col

<Spark Session declaration>

df_final = cast_datatypes(df_target_A, df_source_A)

方法 2:

spark.sparkContext.addPyFile("/Workspace/Shared/Common/common.py")
import common as C
from pyspark.sql.functions import col

df_final = C.cast_datatypes(df_target_A, df_source_A)

这两种方法都能够导入函数，但无法使用“col”。我收到的错误是:

File /Workspace/Shared/Common/common.py:13, in cast_datatypes(df_target, df_source)
     11 types = [f.dataType for f in df_target.schema.fields]
     12 for i, c in enumerate(cols):
---> 13     df_source = df_source.withColumn(c, col(c).cast(types[i]))
     14 df_source = df_source.select(cols)
     15 return df_source

NameError: name 'col' is not defined

我们需要传递导入函数正在使用的所有参数吗？如果是，那么我如何在数据 block 中模块化我的笔记本？

Answer 1

@Reyk 说的是对的。您需要在 common.py 中导入 col。但你仍然遇到错误，即使我也遇到同样的错误。更改 common.py 后，您需要重新启动集群或分离并重新附加笔记本。

/Workspace/Shared/Common/common.py

def cal(df):
    re = df.withColumn("tmp", col("trip_distance") + lit(1))
    return re

当我尝试下面的代码时，出现错误。

import sys
sys.path.append("/Workspace/Shared/Common/")
df = spark.sql("select * from samples.nyctaxi.trips")

from common import cal
from pyspark.sql.functions import col
display(cal(df))

所以我添加了修改后的common.py，如下所示。

from pyspark.sql.functions import col,lit

def cal(df):
    re = df.withColumn("tmp", col("trip_distance") + lit(1))
    return re

再次尝试调用该函数，出现同样的错误。

但是如果我看到文件的内容，它就会被更新。

因此，请执行以下任一操作并重新运行您的代码。

输出:

tmp 列已创建。

如果您再次进行任何更新，则需要重新连接笔记本电脑或重新启动。

如果您不想重新启动或重新附加，则需要重新加载模块common。

更改 common.py 中的代码后，只需使用以下代码。

import sys
import importlib
sys.path.append("/Workspace/Shared/Common/")
df = spark.sql("select * from samples.nyctaxi.trips")
if "common" in sys.modules:
    del sys.modules["common"]

from common import cal
display(cal(df))