我想通过 pyspark sql 在 databricks 上的数据库中查找具有特定列的表。
我使用以下代码，但它不起作用。 https://medium.com/@rajnishkumargarg/find-all-the-tables-by-column-name-in-hive-51caebb94832
在 SQL 服务器上我的代码：
SELECT Table_Name, Column_Name
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_CATALOG = 'YOUR_DATABASE'
AND COLUMN_NAME LIKE '%YOUR_COLUMN%'
但是，我不知道如何在 pyspark sql 上做同样的事情？
谢谢
【问题讨论】：
标签： sql database apache-spark hive pyspark
SparkSession 有一个属性catalog。此目录的方法listTables 返回 SparkSession 已知的所有表的列表。使用此列表，您可以使用 listColumns 查询每个表的所有列
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("test").getOrCreate()
spark.sql("CREATE TABLE tab1 (name STRING, age INT) USING parquet")
spark.sql("CREATE TABLE tab2 (name STRING, age INT) USING parquet")
spark.sql("CREATE TABLE tab3 (street STRING, age INT) USING parquet")
for table in spark.catalog.listTables():
for column in spark.catalog.listColumns(table.name):
if column.name == 'name':
print('Found column {} in table {}'.format(column.name, table.name))
打印
Found column name in table tab1
Found column name in table tab2
如果您想将搜索限制在单个数据库中，listTables 和 listColumns 这两种方法都接受数据库名称作为可选参数。
【讨论】：
如 werner 所述，SparkSession 确实具有目录属性。
如果我对您的理解正确，您希望获得具有特定列的表。 你可以试试这个代码（对不起，scala代码而不是python）：
val databases = spark.catalog.listDatabases().select($"name".as("db_name")).as("databases")
val tables = spark.catalog.listTables().select($"name".as("table_name"), $"database").as("tables")
val tablesWithDatabase = databases.join(tables, $"databases.db_name" === $"tables.database", "inner").collect()
tablesWithDatabase.foreach(row => {
val dbName = row.get(0).asInstanceOf[String]
val tableName = row.get(1).asInstanceOf[String]
val columns = spark.catalog.listColumns(dbName, tableName)
columns.foreach(column=>{
if (column.name == "Your column")
// Do your logic here
null
})
})
注意我正在收集，所以如果你有很多表/数据库，它可能会导致 OOM 错误，我这样做的原因是因为与 listTables 或 listDatabases 方法相比，它可以在没有参数的情况下调用, listColumns 需要获取 dbName 和 tableName，并且它没有任何唯一的 column id 匹配到 table。
所以该列的搜索将在驱动程序本地完成。
希望对您有所帮助。
【讨论】：
我遇到了与 OP 类似的问题，我需要找到与 LIKE 子句匹配的所有列（包括嵌套列）。
我在这里写了一篇关于它的帖子https://medium.com/helmes-people/how-to-view-all-databases-tables-and-columns-in-databricks-9683b12fee10
但是你可以在下面找到完整的代码。
与之前的答案相比，此解决方案的好处是，它可以在您需要使用 OP 编写的 LIKE '%%' 搜索列的情况下工作。此外，它还允许您在嵌套字段中搜索名称。最后，它创建一个类似于 SQL 的视图，类似于 INFORMATION_SCHEMA 视图。
from pyspark.sql.types import StructType
# get field name from schema (recursive for getting nested values)
def get_schema_field_name(field, parent=None):
if type(field.dataType) == StructType:
if parent == None:
prt = field.name
else:
prt = parent+"."+field.name # using dot notation
res = []
for i in field.dataType.fields:
res.append(get_schema_field_name(i, prt))
return res
else:
if parent==None:
res = field.name
else:
res = parent+"."+field.name
return res
# flatten list, from https://stackoverflow.com/a/12472564/4920394
def flatten(S):
if S == []:
return S
if isinstance(S[0], list):
return flatten(S[0]) + flatten(S[1:])
return S[:1] + flatten(S[1:])
# list of databases
db_list = [x[0] for x in spark.sql("SHOW DATABASES").rdd.collect()]
for i in db_list:
spark.sql("SHOW TABLES IN {}".format(i)).createOrReplaceTempView(str(i)+"TablesList")
# create a query for fetching all tables from all databases
union_string = "SELECT database, tableName FROM "
for idx, item in enumerate(db_list):
if idx == 0:
union_string += str(item)+"TablesList WHERE isTemporary = 'false'"
else:
union_string += " UNION ALL SELECT database, tableName FROM {}".format(str(item)+"TablesList WHERE isTemporary = 'false'")
spark.sql(union_string).createOrReplaceTempView("allTables")
# full list = schema, table, column
full_list = []
for i in spark.sql("SELECT * FROM allTables").collect():
table_name = i[0]+"."+i[1]
table_schema = spark.sql("SELECT * FROM {}".format(table_name))
column_list = []
for j in table_schema.schema:
column_list.append(get_schema_field_name(j))
column_list = flatten(column_list)
for k in column_list:
full_list.append([i[0],i[1],k])
spark.createDataFrame(full_list, schema = ['database', 'tableName', 'columnName']).createOrReplaceTempView("allColumns")```
【讨论】：