SparkSession无write属性报错求助：PySpark写入MSSQL问题

运行代码尝试将DataFrame通过JDBC写入MSSQL时，出现错误：AttributeError: 'SparkSession' object has no attribute 'write'

原代码 #

写入函数 #

import pandas as pd
from pyspark.sql import SparkSession
from pyspark import SparkConf


def sql_writer(db, table_name, df, mode):
  server = 'my-server'
  username = 'username'
  password = 'password'
  schema = 'Reports'
  database = db

  sconf=SparkConf().setAppName("test")
  spark=SparkSession.builder.config(conf=sconf).getOrCreate()

    # Write Spark DataFrame to MSSQL using JDBC
  spark.write \
        .format("jdbc") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("user", username) \
        .option("password", password) \
        .mode("append") \
        .save()

  print("DataFrame written to MSSQL successfully!")

调用代码 #

sql_writer('DatabaseName',
          'TableName',
          mydf,
          mode=mode
        )

错误原因及修复方案 #

核心错误 #

write是Spark DataFrame的专属方法，并非SparkSession对象的方法，你错误调用了spark.write，应当使用传入的DataFrame实例来调用write。

除此之外，代码还有几处问题需要修正：

1. 未定义url变量：需拼接MSSQL的JDBC连接URL，格式为jdbc:sqlserver://{server}:1433;databaseName={database};schema={schema}（端口默认1433，若你的MSSQL端口不同请修改）。
2. 未使用传入的mode参数：代码中硬编码了mode("append")，应替换为函数参数mode。
3. Pandas DataFrame转Spark DataFrame（可选）：如果传入的mydf是Pandas DataFrame，需先通过spark.createDataFrame(mydf)转换为Spark DataFrame，否则无法使用Spark的JDBC写入方法。

修复后的完整代码 #

import pandas as pd
from pyspark.sql import SparkSession
from pyspark import SparkConf


def sql_writer(db, table_name, df, mode):
    server = 'my-server'
    username = 'username'
    password = 'password'
    schema = 'Reports'
    database = db

    # 拼接JDBC URL
    url = f"jdbc:sqlserver://{server}:1433;databaseName={database};schema={schema}"

    sconf = SparkConf().setAppName("test")
    spark = SparkSession.builder.config(conf=sconf).getOrCreate()

    # 如果传入的是Pandas DataFrame，转换为Spark DataFrame
    if isinstance(df, pd.DataFrame):
        df = spark.createDataFrame(df)

    # 使用Spark DataFrame调用write方法
    df.write \
        .format("jdbc") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("user", username) \
        .option("password", password) \
        .mode(mode) \
        .save()

    print("DataFrame written to MSSQL successfully!")

调用示例 #

# 假设mode为'append'或'overwrite'等合法写入模式
sql_writer('DatabaseName',
           'TableName',
           mydf,
           mode='append'
           )

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