如何处理 Spark SQL 上的 AnalysisException？

Question

我正在尝试在 Spark 中执行查询列表，但如果查询未正确运行，Spark 会向我抛出以下错误： AnalysisException: “ALTER TABLE CHANGE COLUMN is not supported for changes ...

这是我的代码的一部分（我在 Databricks 上使用 python 和 Spark SQL）：

for index, row in df_tables.iterrows():
  query = row["query"]
  print ("Executing query: ")
  try:
      spark.sql(query)
      print ("Query executed")
  except (ValueError, RuntimeError, TypeError, NameError):
      print("Unable to process your query dude!!")
  else:
      #do another thing

有什么办法可以捕获该异常吗？ValueError、RuntimeError、TypeError、NameError 似乎不起作用。Spark 网页中没有太多相关信息。

Answer 1

我发现在 pyspark.sql.utils 中定义了 AnalysisException。 https://spark.apache.org/docs/3.0.1/api/python/_modules/pyspark/sql/utils.html

import pyspark.sql.utils
try:
    spark.sql(query)
    print ("Query executed")
except pyspark.sql.utils.AnalysisException:
    print("Unable to process your query dude!!")

Answer 2

您可以修改 try except 语句，如下所示：

try:
  spark.sql(query)
  print ("Query executed")
except Exception as x:
  print("Unable to process your query dude!!" + \
        "\n" + "ERROR : " + str(x)) 

Answer 3

我想提出一种选择特定异常的方法。我在查找某些表是否已经存在时遇到问题。我发现的最简单的方法就是这样。当然，如果 Spark 维护者更改异常消息，这可能会中断，但我认为在这种情况下他们没有理由这样做。

    import pyspark.sql.utils

    try:
        spark.read.parquet(SOMEPATH)
    except pyspark.sql.utils.AnalysisException as e:
        if "Path does not exist:" in str(e):
            # Finding specific message of Exception.
            pass # run some code to address this specific case.
        else:
            # if this is not the AnalysisException that i was waiting,
            # i throw again the exception
            raise (e)
    except Exception as e:
        # if is another exception i can catch like this
        print(e)
        raise (e)