使用PySpark查询Azure SynapseSQL时含受限关键字列名报错

问题描述 #

使用PySpark查询Azure Synapse Analytics SQL数据库时，列别名包含受限关键字片段会触发报错。例如将c.Name别名为ClosedByName时，因别名包含受限关键字"close"导致失败，但直接使用原列名或别名改为losedByName可正常运行。由于需关联两张含受限关键字列的表，无法通过修改原列名规避问题。

示例代码 #

query = """Select c.Name as ClosedByName, c.Profile_Name__c FROM dbo.table as c"""

# Read from a query
dfToReadFromQueryAsArgument = (spark.read
                     .option(Constants.DATABASE, "server")
                     .option(Constants.SERVER, "workspace.sql.azuresynapse.net")
                     .synapsesql(query)
)
dfToReadFromQueryAsArgument.show()

报错信息 #

Py4JJavaError                             Traceback (most recent call last)
/tmp/ipykernel_6968/4240601845.py in <module>
      1 query = """Select c.Name as ClosedByName,       c.Profile_Name__c FROM dbo.table as c"""
      2 
----> 3 dfToReadFromQueryAsOption = (spark.read
      4                      # Name of the SQL Dedicated Pool or database where to run the query
      5                      # Database can be specified as a Spark Config - spark.sqlanalyticsconnector.dw.database or as a Constant - Constants.DATABASE

~/cluster-env/env/lib/python3.8/site-packages/com/microsoft/spark/sqlanalytics/SqlAnalyticsReader.py in synapsesql(self, table_name)
     40         df = DataFrame(jdf, sqlcontext)
     41     except Exception as e:
---> 42         raise e
     43     return df

~/cluster-env/env/lib/python3.8/site-packages/com/microsoft/spark/sqlanalytics/SqlAnalyticsReader.py in synapsesql(self, table_name)
     37         connector = sqlcontext._jvm.com.microsoft.spark.sqlanalytics.SqlAnalyticsConnectorClass() \
     38             .SQLAnalyticsFormatReader(self._jreader)
---> 39         jdf = connector.synapsesql(table_name)
     40         df = DataFrame(jdf, sqlcontext)
     41     except Exception as e:

~/cluster-env/env/lib/python3.8/site-packages/py4j/java_gateway.py in __call__(self, *args)
   1302 
   1303         answer = self.gateway_client.send_command(command)
-> 1304         return_value = get_return_value(
   1305             answer, self.gateway_client, self.target_id, self.name)
   1306 

/opt/spark/python/lib/pyspark.zip/pyspark/sql/utils.py in deco(*a, **kw)
    109     def deco(*a, **kw):
    110         try:
--> 111             return f(*a, **kw)
    112         except py4j.protocol.Py4JJavaError as e:
    113             converted = convert_exception(e.java_exception)

~/cluster-env/env/lib/python3.8/site-packages/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    324             value = OUTPUT_CONVERTER[type](answer[2:], gateway_client)
    325             if answer[1] == REFERENCE_TYPE:
--> 326                 raise Py4JJavaError(
    327                     "An error occurred while calling {0}{1}{2}.\n".
    328                     format(target_id, ".", name), value)

Py4JJavaError: An error occurred while calling o3616.synapsesql.
: com.microsoft.spark.sqlanalytics.SQLAnalyticsConnectorException: Queries with keywords: 
create
alter
drop
with
exec
execute
insert
delete
disable
enable
update
merge
truncate
backup
restore
collate
close
deny
grant
open
revoke
revert are not allowed
    at com.microsoft.spark.sqlanalytics.utils.SQLAnalyticsConnectorOptionsValidator$.validateOptions(SQLAnalyticsConnectorOptionsValidator.scala:118)
    at com.microsoft.spark.sqlanalytics.utils.SQLAnalyticsConnectorOptionsValidator$.validateOptions(SQLAnalyticsConnectorOptionsValidator.scala:68)
    at com.microsoft.spark.sqlanalytics.utils.Utils$.initializeAndValidateOptions(Utils.scala:122)
    at com.microsoft.spark.sqlanalytics.ItemsTable.readSchema(ItemsTable.scala:96)
    at com.microsoft.spark.sqlanalytics.ItemsTable.$anonfun$schema$1(ItemsTable.scala:88)
    at scala.Option.getOrElse(Option.scala:189)
    at com.microsoft.spark.sqlanalytics.ItemsTable.schema(ItemsTable.scala:88)
    at com.microsoft.spark.sqlanalytics.SynapseSqlDataSourceV2.inferSchema(SynapseSqlDataSourceV2.scala:46)
    at org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.getTableFromProvider(DataSourceV2Utils.scala:81)
    at org.apache.spark.sql.DataFrameReader.$anonfun$load$1(DataFrameReader.scala:303)
    at scala.Option.map(Option.scala:230)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:273)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227)
    at com.microsoft.spark.sqlanalytics.SqlAnalyticsConnectorClass$SQLAnalyticsFormatReader.sqlanalytics(SqlAnalyticsConnectorClass.scala:105)
    at com.microsoft.spark.sqlanalytics.SqlAnalyticsConnectorClass$SQLAnalyticsFormatReader.synapsesql(SqlAnalyticsConnectorClass.scala:82)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:750)

解决方案 #

方法1：Spark端重命名列（规避连接器校验） #

先执行不含受限别名的查询，获取DataFrame后在Spark内部修改列名：

# 查询时使用原始列名
query = """Select c.Name, c.Profile_Name__c FROM dbo.table as c"""

dfToReadFromQueryAsArgument = (spark.read
                     .option(Constants.DATABASE, "server")
                     .option(Constants.SERVER, "workspace.sql.azuresynapse.net")
                     .synapsesql(query)
)

# 在Spark中重命名列
df_renamed = dfToReadFromQueryAsArgument.withColumnRenamed("Name", "ClosedByName")
df_renamed.show()

方法2：用Synapse存储过程封装查询 #

在Synapse中创建存储过程，内部定义带受限别名的查询，PySpark调用存储过程：

1. 在Synapse执行创建存储过程：

CREATE PROCEDURE GetClosedByNameData
AS
BEGIN
    SELECT c.Name AS ClosedByName, c.Profile_Name__c FROM dbo.table AS c;
END

2. PySpark调用存储过程：

query = """EXEC GetClosedByNameData"""

dfToReadFromQueryAsArgument = (spark.read
                     .option(Constants.DATABASE, "server")
                     .option(Constants.SERVER, "workspace.sql.azuresynapse.net")
                     .synapsesql(query)
)
dfToReadFromQueryAsArgument.show()

方法3：用Synapse视图封装查询 #

在Synapse中创建视图，将别名定义在视图内，PySpark直接查询视图：

1. 创建视图：

CREATE VIEW vw_ClosedByNameData
AS
SELECT c.Name AS ClosedByName, c.Profile_Name__c FROM dbo.table AS c;

2. PySpark查询视图：

query = """SELECT * FROM dbo.vw_ClosedByNameData"""

dfToReadFromQueryAsArgument = (spark.read
                     .option(Constants.DATABASE, "server")
                     .option(Constants.SERVER, "workspace.sql.azuresynapse.net")
                     .synapsesql(query)
)
dfToReadFromQueryAsArgument.show()

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