在 Python 工作表中编写 Snowpark 代码¶

查看并接受 Anaconda 服务条款¶

在开始使用 Snowflake 中的 Anaconda 提供的包之前，您必须确认 外部产品条款。

1. 登录 Snowsight。

2. 选择 Admin » Billing & Terms。

3. 在 Anaconda 部分中，选择 Enable。

4. 在 Anaconda Packages 对话框中，单击链接查看 外部产品条款页面。

5. 如果您同意条款，请选择 Acknowledge & Continue。

如果在尝试接受服务条款时出现错误，您的用户简介中可能缺少名、姓或电子邮件地址。如果您有管理员角色，请参阅 向您的用户简介添加用户详细信息，以使用 Snowsight 更新您的简介。否则，请联系管理员 更新您的账户。

将 Python 文件从暂存区添加到工作表¶

Snowflake 在 Python 工作表中包含来自 ` Snowflake Anaconda 频道 <https://repo.anaconda.com/pkgs/snowflake/ (https://repo.anaconda.com/pkgs/snowflake/)>`_ 的 Snowpark 包。

如果要在 Python 工作表中使用 Python 文件或包，而不是 Anaconda 中包含的文件或包，则必须将这些文件上传到 Snowflake 中的命名暂存区，然后为 Python 工作表将它们添加到已安装包列表中。

若要在工作表中使用 Python 包，请执行以下操作：

1. 登录 Snowsight。

2. 打开 Projects » Worksheets。

3. 选择 + » Python Worksheet。

4. 选择数据库和架构。

5. 选择 Packages » Stage Packages。

6. 在暂存区中输入包的路径：

   • 如果工作表的选定数据库和架构包含包所在的暂存区，则可以使用非限定名称引用该暂存区。例如 @YourStage/path/to/example_package.py。

   • 若要引用其他数据库和架构中的暂存区，请完全限定暂存区的名称。例如 @Database.Schema.Stage/path/to/other_package.py。

7. 选择 Import，可将包添加到已安装包的列表。

8. 在代码中，使用 import 语句在 Python 工作表中使用 包。例如，从 example_package.py 和 other_package.py 文件导入包后，编写以下代码以从 example_package 导入名为 function 的函数，并导入包 other_package 以便在代码中使用：

   from example_package import function
   import other_package
   

   Copy

有关更多详细信息，请参阅 为代码提供依赖项。

返回不同数据类型的结果¶

编写 Python 代码时，请考虑代码中的 return 语句返回哪种类型的数据，并调整工作表返回结果的方式。默认情况下，Python 工作表有一个 Table() 的返回类型，因为占位符代码返回 DataFrame。

根据 Python 代码返回的内容，您可能需要更改工作表设置以用不同的方式显示输出：

• 如果处理程序函数返回 DataFrame，请使用 Table() 的默认返回类型。

• 如果处理程序函数返回 Row 对象列表（例如使用 collect 方法），请将返回类型更改为 Variant。

• 如果处理程序函数返回字符串（如 return "Hello Python"）或返回要类型转换为字符串的值，请将返回类型更改为 String。

• 如果处理程序函数返回整型（例如使用 count 方法），请使用 Variant 或 String 的返回类型。

有关某些 DataFrame 方法的返回类型的详细信息，请参见 执行操作以计算 DataFrame。

若要更新工作表设置以返回不同类型的结果，请执行以下操作：

1. 登录 Snowsight。

2. 打开 Projects » Worksheets。

3. 打开要将结果显示为表的 Python 工作表。

4. 选择要用于运行工作表的仓库。如果您有一个用户的默认仓库，则该仓库是预先选择的。确保您的仓库正在运行。

5. 选择 Settings 并为 Return type 选择处理程序函数返回的类型。

6. 运行 Python 工作表。

7. 在 Results 面板中查看结果。

将其他实参传递给处理程序函数¶

使用 Python 工作表，您可以通过将该函数指定为工作表的处理程序来测试采用单个实参（Snowpark Session 对象）的 Python 函数。Python 工作表中定义的每个函数都需要传入 session: snowpark.Session 实参。

若要测试传入其他实参的函数，请执行以下操作：

1. 将实参添加到函数中。

2. 定义一个单独的单实参函数，该函数传入 Snowpark Session。在此函数中，调用多实参函数，传入其他实参的值，然后返回函数的值。

   例如，要编写按包语言列筛选包表的 Snowpark Python 代码，您可以编写以下代码：

   import snowflake.snowpark as snowpark
   from snowflake.snowpark.functions import col
   
   # Add parameters with optional type hints to the main handler function
   def main(session: snowpark.Session, language: str):
     # Your code goes here, inside the "main" handler.
     table_name = 'information_schema.packages'
     dataFrame = session.table(table_name).filter(col("language") == language)
   
     # Print a sample of the dataFrame to standard output
     dataFrame.show()
   
     # The return value appears in the Results tab
     return dataFrame
   
   # Add a second function to supply a value for the language parameter to validate that your main handler function runs.
   def test_language(session: snowpark.Session):
     return main(session, 'java')
   

   Copy

   在此示例中， main 函数是多实参函数， test_language 函数是单实参函数，用于验证代码是否使用已传递的实参值运行。

3. 将单实参函数设置为处理程序函数以运行工作表，并验证代码是否使用实参值运行。

   在此示例中，将处理程序更改为 test_language 函数，然后选择 Run。您可以在工作表 Settings 中更改处理程序，或选择处理程序函数旁边的 Show actions Lightbulb，然后选择 Set function "test_language" as handler。

:ref:` 将 Python 工作表部署为存储过程 <label-snowsight_py_worksheets_deploy_stored_proc>` 时，可以选择主处理程序函数并查看存储过程的实参和映射类型。

查看代码生成的输出¶

您可以在 Python 工作表的 Output 面板中查看 Python 代码的标准输出 (stdout) 或标准错误 (stderr) 消息。

您可以在 Output 面板中看到以下类型函数的输出：

• 写入控制台的函数，例如 print()。

• 打印 DataFrame 的函数，例如 Snowpark Python 中 DataFrame 类的 show 方法。

查看 Python 工作表的查询历史记录¶

当 Python 工作表在 Snowsight 中运行时，匿名存储过程将运行代码并生成在代码中执行 Snowpark 命令的查询。

您可以使用 Snowsight 中的 Query History 页面来查看运行的查询。请参阅 使用 Snowsight 查看 Query History。

例如，运行工作表后，可以通过执行以下操作来查看运行的查询：

1. 查看工作表的 Results。

2. 在工作表中的 Query Details，选择 » Copy Query ID

3. 选择 Projects » Worksheets 可返回到工作表列表。

4. 选择 Monitoring » Query History。

5. 在 Query History 页面上，仅显示 Python 工作表中的查询：

   1. 选择 Filters，然后启用 Query ID 选项。

   2. 输入 Python 工作表的查询 ID。

   3. 选择 Apply Filters。

6. 查看工作表的查询运行。

示例：编写一个简单的 Snowpark 程序¶

在此示例中，编写一个 Snowpark Python 程序，该程序生成一个小范围的数字，并将该范围写入代码在 Snowflake 中创建的表，或者替换该表（如果已存在）。若要运行此代码示例，您必须对要添加表的数据库架构具有 CREATE TABLE 权限。

1. 登录 Snowsight。

2. 打开 Projects » Worksheets。

3. 选择 + » Python Worksheet。

4. 选择要将表添加到其中的数据库和架构。

5. 选择要用于运行工作表的仓库。如果您有一个用户的默认仓库，则该仓库是预先选择的。确保您的仓库正在运行。

6. 将 Snowpark Python 代码编写为 main 函数的一部分：

   import snowflake.snowpark as snowpark
   
   def main(session: snowpark.Session):
     tableName = "range_table"
     df_range = session.range(1, 10, 2).to_df('a')
     df_range.write.mode("overwrite").save_as_table(tableName)
     return tableName + " table successfully created"
   

   Copy

7. 选择 Settings 并为 Return type 选择处理程序函数所返回类型的 String。

8. 运行代码。

示例：在 Python 工作表中转换数据¶

在此示例中，编写 Python 代码，该代码按计划时间和状态在 SNOWFLAKE 数据库的 ACCOUNT_USAGE 架构中汇总 TASK_HISTORY 视图中的条目，并将汇总输出保存到表 aggregate_task_history。

1. 登录 Snowsight。

2. 打开 Projects » Worksheets。

3. 选择 + » Python Worksheet。

4. 选择要将表添加到其中的数据库和架构。

5. 选择要用于运行工作表的仓库。如果您有一个用户的默认仓库，则该仓库是预先选择的。确保您的仓库正在运行。

6. 将 Snowpark Python 代码编写为 main 函数的一部分：

   import snowflake.snowpark as snowpark
   from snowflake.snowpark.functions import col
   from snowflake.snowpark.dataframe_reader import *
   from snowflake.snowpark.functions import *
   
   def main(session: snowpark.Session):
   
     inputTableName = "snowflake.account_usage.task_history"
     outputTableName = "aggregate_task_history"
   
     df = session.table(inputTableName)
     df.filter(col("STATE") != "SKIPPED")\
       .group_by(("SCHEDULED_TIME"), "STATE").count()\
       .write.mode("overwrite").save_as_table(outputTableName)
     return outputTableName + " table successfully written"
   

   Copy

7. 选择 Settings 并为 Return type 选择处理程序函数所返回类型的 String。

8. 运行代码。

在 Python 工作表中运行代码后，可以打开 SQL 工作表并查询表。请参阅 使用工作表查询数据。

示例：使用 Python 工作表从暂存区读取文件¶

Snowpark Python 允许您从暂存区读取文件并将内容写入表或将它们保存为 Snowflake 中的视图。在此示例中，Python 代码从 db1.public.files 命名暂存区读取 CSV 格式的已压缩文件（包含员工数据 data_0_0_0.csv.gz ）的内容，并将内容写入名为 employees 的表中。

1. 登录 Snowsight。

2. 打开 Projects » Worksheets。

3. 选择 + » Python Worksheet。

4. 选择要将表添加到其中的数据库和架构。

5. 选择要用于运行工作表的仓库。如果您有一个用户的默认仓库，则该仓库是预先选择的。确保您的仓库正在运行。

6. 将 Snowpark Python 代码编写为 main 函数的一部分：

   import snowflake.snowpark as snowpark
   from snowflake.snowpark.types import *
   
   schema_for_file = StructType([
     StructField("name", StringType()),
     StructField("role", StringType())
   ])
   
   fileLocation = "@DB1.PUBLIC.FILES/data_0_0_0.csv.gz"
   outputTableName = "employees"
   
   def main(session: snowpark.Session):
     df_reader = session.read.schema(schema_for_file)
     df = df_reader.csv(fileLocation)
     df.write.mode("overwrite").save_as_table(outputTableName)
   
     return outputTableName + " table successfully written from stage"
   

   Copy

7. 选择 Settings 并为 Return type 选择处理程序函数所返回类型的 String。

8. 运行代码。

在 Python 工作表中运行代码后，可以打开 SQL 工作表并查询表。请参阅 使用工作表查询数据。

有关使用 Snowpark 在暂存区上处理文件的更多详细信息，请参阅 在 Snowpark Python 中使用 DataFrames。