如果使用pyodbc，catalyst优化器生成的java字节码只作为一个节点（executor）运行，而不是整个集群。对于更大的数据集，这会妨碍集群的充分利用和性能问题。

最好使用JDBC的spark驱动程序，微软有一个。
https://learn.microsoft.com/en-us/sql/connect/spark/connector?view=sql-server-ver16
将复杂的T-SQL创建为视图，然后读取它们。这就是Spark为阅读文件所做的。使用JDBC驱动程序（spark），如果需要，它将允许您通过更改分区方法来并行读取。
为正确版本的spark安装Marven库。

我使用的是Spark版本〉3.1。
我有一个冒险作品数据库，其中有一个视图叫v。

#
#  Set connection properties
#

server_name = "jdbc:sqlserver://svr4tips2030.database.windows.net"
database_name = "dbs4advwrks"
url = server_name + ";" + "databaseName=" + database_name + ";"
table_name = "dbo.vDMPrep"
user_name = "enter your user here"
password = "enter your password here"

使用JDBC驱动程序进行典型的spark.read（）调用。

df = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("user", user_name) \
        .option("password", password).load()

display(df)

以下是显示 Dataframe 的结果。

数据框是严格类型化的吗？答案是肯定的，因为它从SQL Server获取字段信息。
x1c4d 1x指令集
最后但并非最不重要的一点是，视图是否复杂？下图显示了8个表被联接和聚合以获得视图的最终结果。

总之，使用数据库中的视图为Spark预编译数据集。使用Microsoft的JDBC驱动程序通过dataframe读取和写入SQL Server。
至于存储过程，有一种方法可以使用驱动程序来执行非查询。我将不得不寻找代码。请继续关注更新或第2部分。

这是答案的第二部分。没有好的方法将存储过程调用的结果作为 Dataframe 返回。
以下是MSFT github站点上该驱动程序的链接，声明不支持存储过程。
https://github.com/microsoft/sql-spark-connector/issues/21
这是一个黑客的工作。
在我的例子中，我的SP要做一些工作，并将其保存到一个临时表中。然后使用上述技术读取该表。
下面的代码删除表（如果存在），然后重新加载。

-- 
-- Sample Call
-- 
CREATE PROCEDURE dbo.StackOverFlowTest
AS
BEGIN
    DROP TABLE IF EXISTS stage.DimSalesTerritory;
    SELECT * INTO stage.DimSalesTerritory FROM dbo.DimSalesTerritory
END

下面是进入底层JAVA驱动程序管理器的代码。它有一个调用SP的属性。

#
#  Grab the low level driver manager, exec sp
#

driver_manager = spark._sc._gateway.jvm.java.sql.DriverManager
connection = driver_manager.getConnection(url, user_name, password)
connection.prepareCall("EXEC dbo.StackOverFlowTest").execute()
connection.close()

使用spark.read（）从SP填充的新表中检索数据。

我希望这适合你的用例。同样，这将不会伸缩，因为它运行在控制节点（执行器）。如果你有一个5节点集群，这将只运行在1个节点。
也就是说，我们可以让spark推断数据类型。如果你看pyodbc文档，你必须安装一个本地ODBC驱动程序。这对工作站上的anaconda是好的，但对spark集群是坏的。相反，用户pymssql模块是自包含的本机代码。使用集群库的PyPi部分安装。
https://learn.microsoft.com/en-us/sql/connect/python/pymssql/python-sql-driver-pymssql?view=sql-server-ver16
现在我们有了一个驱动程序，让我们编写一个模块，它将从SELECT语句或存储过程调用EXEC返回一个 Dataframe 。

#
#  Create function to call tsql + return df
#

# use module
import pymssql  

# define function
def call_tsql(info):
  
  # make connection
  conn = pymssql.connect(server=info[0], user=info[1], password=info[2], database=info[3])  
  
  # open cursor
  cursor = conn.cursor()  
  cursor.execute(info[4])
  
  # grab column data (name, type, ...)
  desc = cursor.description
  
  # grab data as list of tuples
  dat1 = cursor.fetchall()
  
  # close cursor
  conn.commit()
  conn.close()
  
  # extract column names
  col1 = list(map(lambda x: x[0], desc))
  
  # let spark infer data types
  df1 = spark.createDataFrame(data=dat1, schema=col1)
  
  # return dataframe
  return df1

此代码将只支持一个结果集。如果需要MARS，请修改多个活动结果集。
有大量的注解。简而言之，连接和TSQL的信息元组作为参数传入，并返回 Dataframe 。
是的，cursor.description中有数据类型，但它们是编码的。我没有找到合适的Map。由于您不处理大型数据，请推断架构。否则，请为架构传递DDL语句而不是列标题。

#
# Make call using SELECT statement
#

# tuple of info (server, user, pwd, database, query)
info = ('svr4tips2030.database.windows.net', '<your user>', '<your pwd>', 'dbs4advwrks', 'select * from dbo.vDMPrep')

# get data frame
df2 = call_tsql(info)

上图显示了推断的类型，下图显示了数据。

我创建了一个非常简单的存储过程，它只从视图中进行选择。

CREATE PROCEDURE [dbo].[StackOverFlowTest]
AS
BEGIN
    SELECT * FROM [dbo].[vDMPrep]
END
GO

如果我们更改调用以执行此存储过程，则会得到相同的答案。

#
# Make call using EXEC statement
#

# tuple of info (server, user, pwd, database, query)
info = ('svr4tips2030.database.windows.net', '<your user>', '<your pwd>', 'dbs4advwrks', 'exec [dbo].[StackOverFlowTest]')

# get data frame
df2 = call_tsql(info)

随附的是一个示例SPARKDDL语句。请参见JSON文档的file_schema部分，它作为参数传递给笔记本。

#
# Table 1 - dim.account
#

# Set parameters for notebook
parms = {
"datalake_path": "/mnt/datalake/bronze/",
"datafile_path": "/dim/account/dim-account-20200905T101530.csv",
"debug_flag": "false",
"partition_count": "2",
"file_schema": "AccountKey INT, ParentAccountKey INT, AccountCodeAlternateKey INT, ParentAccountCodeAlternateKey INT, AccountDescription STRING, AccountType STRING, Operator STRING, CustomMembers STRING, ValueType STRING, CustomMemberOptions STRING"
}

# Run notebook with selections
ret = dbutils.notebook.run("./nb-full-load-delta-table", 60, parms)

# Show return value if any
print(ret)

概括地说，如果您遇到性能问题，请创建一个单节点集群并在该节点上扩展计算。不可以，因为pymssql模块不会使用多个节点。