如何将Azure Synapse中的PySpark Notebook转换为PySpark作业

完全可以把Notebook里的DataFrame转换逻辑迁移到.py文件并创建PySpark作业定义，但要注意以下几个关键细节：

• 彻底处理display()语句：display()会触发DataFrame的collect()操作，把数据拉到前端展示，这是Notebook运行慢的核心原因。如果需要留存结果，直接把数据写入Synapse Lakehouse、Blob存储或者SQL池即可，比如用df.write.mode("overwrite").parquet("abfss://container@storage.dfs.core.windows.net/path")或者df.write.mode("append").saveAsTable("target_db.target_table")；如果是调试阶段需要看样本数据，临时用df.limit(100).show()也可以，但生产作业里一定要删掉这类操作。

• 显式配置上下文依赖：Notebook里很多配置是隐含的（比如挂载的存储、会话参数、第三方库），迁移到.py文件要全部显式写出来：

  • 会话参数：比如设置 shuffle 分区数spark.conf.set("spark.sql.shuffle.partitions", "200")
  • 第三方库：如果用到了非默认库，要先在Synapse库管理上传依赖包，然后在作业配置里勾选对应的库

• 结构化代码逻辑：把Notebook里零散的单元格代码整理成清晰的函数或流程，比如拆分数据读取、转换、写入三个模块，方便维护和调试。示例代码：

  from pyspark.sql import SparkSession
  
  def read_source(spark):
      return spark.read.parquet("abfss://source_path")
  
  def transform_data(raw_df):
      return raw_df.filter("status = 'valid'").join(dim_df, "id", "inner")
  
  def write_result(processed_df):
      processed_df.write.mode("overwrite").saveAsTable("dw.fact_table")
  
  if __name__ == "__main__":
      spark = SparkSession.builder.appName("Synapse_ETL_Job").getOrCreate()
      raw_data = read_source(spark)
      processed_data = transform_data(raw_data)
      write_result(processed_data)
      spark.stop()
  

• 合理配置作业资源：创建PySpark作业时，要根据数据量调整集群规格（节点类型、节点数），避免资源不足拖慢速度或者资源浪费。同时设置好作业超时时间、重试策略，指定日志存储路径，方便后续排查问题。

• 处理交互式参数：如果Notebook里有用到手动输入的变量（比如日期、批次号），在.py文件里改成通过命令行参数传递，比如用sys.argv接收：

  import sys
  batch_date = sys.argv[1]
  filtered_df = raw_df.filter(f"dt = '{batch_date}'")
  

  然后在作业定义的「参数」栏里传入具体值即可。

• 先调试再上线：迁移完成后，先用小数据集测试.py脚本的逻辑正确性，比如在Synapse里用小集群提交临时作业，或者在Notebook里导入.py的函数做局部验证，没问题再跑生产级别的作业。

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