CDAP流水线中PySpark脚本报错：NameError: name 'SparkSession' is not defined

我刚看到你的问题，这种本地跑通但CDAP里报错的情况很常见，主要是因为CDAP的PySpark运行上下文和本地独立Spark脚本的环境不一样。咱们一步步来解决：

核心原因 #

你在本地命令行运行的是独立PySpark程序，可以自由初始化SparkSession；但在CDAP流水线中，Spark的上下文是由CDAP托管管理的，它不会让你随便自己创建Session，而是需要通过CDAP提供的context对象来获取已初始化好的SparkSession。另外，你用的from pyspark.sql import *通配符导入，在CDAP的环境里可能没有正确引入SparkSession，导致报错。

修复步骤 #

1. 显式导入SparkSession
   把通配符导入改成明确导入SparkSession，避免环境差异导致的导入问题：

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import trim, to_date, year, month
   

2. 适配CDAP的脚本结构
   CDAP的PySpark作业要求脚本定义一个run函数，接收context参数——这个context是CDAP传递给你的，里面包含了所有Spark相关的上下文信息。你需要从这里获取SparkSession，而不是自己创建：

   def run(context):
       # 从CDAP的context中获取已初始化的SparkSession
       spark = context.getSparkSession()
       
       # 下面写你原本的业务逻辑，比如读取数据、处理转换等
       # 示例：
       # df = spark.read.csv("your-input-path")
       # df = df.withColumn("clean_date", to_date(trim(df.raw_date)))
       # ...
   

3. 额外排查点（如果还是报错）

   • 检查CDAP集群使用的Spark版本：SparkSession是在Spark 2.0之后才引入的，如果你的CDAP集群用的是Spark 1.x，那得改用SQLContext代替，代码调整成：

     from pyspark.sql import SQLContext
     
     def run(context):
         sql_context = SQLContext(context.getSparkContext())
         # 后续用sql_context操作数据
     

   • 确认CDAP流水线的PySpark节点配置是否正确：比如Spark版本选择、资源分配等，确保和你本地测试的环境版本匹配。

这样调整后，你的脚本应该就能在CDAP流水线里正常运行了。

内容的提问来源于stack exchange，提问作者Matteo Perico