获取trial_db数据库所有表列名时遇AnalysisException错误求助

需求：列出trial_db数据库下所有表的列名，已成功获取表名，但添加列名逻辑时触发如下错误：

AnalysisException: Table or view not found: countrycurrency_csv; line 1 pos 9;
---------------------------------------------------------------------------
AnalysisException                         Traceback (most recent call last)
<command-1607350> in <module>
     16   # For each table, get list of columns
     17   for table in tables:
---> 18     columns = [c.name for c in spark.sql(f"describe {table}").collect()]
     19     # Create a dataframe with database, table, and columns information
     20     df = spark.createDataFrame([(db.databaseName, table, columns)], schema=['database', 'table', 'columns'])

/databricks/spark/python/pyspark/sql/session.py in sql(self, sqlQuery)
    775         [Row(f1=1, f2='row1'), Row(f1=2, f2='row2'), Row(f1=3, f2='row3')]
    776         """
---> 777         return DataFrame(self._jsparkSession.sql(sqlQuery), self._wrapped)
    778 
    779     def table(self, tableName):

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/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)

错误原因分析 #

1. 表引用不完整：执行describe {table}时未指定数据库名，Spark默认在当前会话的数据库中查找表，若当前数据库不是trial_db则会找不到目标表。
2. 列名取值错误：spark.sql("describe table")返回的结果中，列名字段是col_name而非name，原代码中c.name会导致取值异常。
3. DataFrame结构不匹配：初始的tbl_df由show tables生成，结构为database, tableName, isTemporary，而后续创建的DataFrame结构是database, table, columns，union操作会因结构不一致报错。
4. 无效函数冗余：create_df函数未被调用，且内部引用未定义的df对象，同时String应改为StringType()（需导入对应类型模块）。

修正后的代码 #

from pyspark.sql.types import StructType, StructField, StringType, ArrayType

# 初始化空DataFrame，与最终结果结构保持一致
result_schema = StructType([
    StructField("database", StringType(), True),
    StructField("table", StringType(), True),
    StructField("columns", ArrayType(StringType()), True)
])
tbl_df = spark.createDataFrame([], schema=result_schema)

# 遍历指定数据库
for db in spark.sql("show databases like 'trial_db'").collect():
    db_name = db.databaseName
    # 获取当前数据库下的所有表名
    tables = spark.sql(f"show tables in {db_name}").rdd.map(lambda row: row.tableName).collect()
    
    # 逐个表获取列名
    for table in tables:
        # 使用完整限定名定位表，避免找不到表的问题
        desc_result = spark.sql(f"describe {db_name}.{table}")
        # 提取真实列名，过滤describe返回的元信息行（以#开头）
        column_list = [row.col_name for row in desc_result.collect() if not row.col_name.startswith("#")]
        # 创建单条记录的DataFrame
        single_row_df = spark.createDataFrame([(db_name, table, column_list)], schema=result_schema)
        # 合并到主结果DataFrame
        tbl_df = tbl_df.union(single_row_df)

# 展示最终结果（truncate=False完整显示列名列表）
tbl_df.show(truncate=False)

关键修正点说明 #

• 用{db_name}.{table}完整限定名引用表，确保Spark能准确定位到目标数据库下的表。
• 初始化时定义统一的Schema，避免union操作的结构不匹配问题。
• 过滤describe结果中的元信息行，只保留真实列名。
• 移除冗余的无效函数，简化代码逻辑。

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