如何在AWS Glue Python环境中搜索打印元数据列名并查询同名列的表

Hey there! Let's build exactly what you need using the AWS Glue Python API in your development endpoint. I'll break this down into clear, actionable parts first—replicating those Pandas-style database/table access patterns, then building the column-search function you're after.

1. Replicate Pandas-style Database & Table Access #

First, let's get that dbs = awsGlue.databases and tables_db1 = dbs[0].tables behavior working. We'll use the boto3 client for Glue (which is pre-available in your development endpoint):

Initialize Glue Client #

import boto3

# Create a Glue client instance to interact with the Data Catalog
glue_client = boto3.client('glue')

Get All Databases (equivalent to dbs = awsGlue.databases) #

This gives you a clean list of all databases in your Glue Data Catalog:

# Fetch all databases from the catalog
db_response = glue_client.get_databases()
# If you just want database names (like a simple list):
dbs = [db['Name'] for db in db_response['DatabaseList']]
# If you need full metadata (like descriptions or locations):
# dbs = db_response['DatabaseList']

Get Tables in a Specific Database (equivalent to tables_db1 = dbs[0].tables) #

Grab all tables from the first database in your list:

# Target the first database in your dbs list
target_db = dbs[0]
# Fetch all tables in this database
tables_response = glue_client.get_tables(DatabaseName=target_db)
# Get just table names:
tables_db1 = [table['Name'] for table in tables_response['TableList']]
# Get full table details (including schema and storage info):
# tables_db1 = tables_response['TableList']

2. Build the Function to Find Tables with a Specific Column #

Now let's create the core function that scans your catalog to find all tables containing your target column, along with their column details (just like df.columns but across all your tables):

The Search Function #

def find_tables_with_column(target_column, target_database=None):
    """
    Search the Glue Data Catalog for tables containing a specific column.
    Returns table metadata and matching column details.
    
    Args:
        target_column (str): Column name to search for (case-insensitive by default)
        target_database (str, optional): Limit search to a single database. 
                                        Leave None to search all databases.
    
    Returns:
        list: List of dictionaries with database name, table name, and matching column info
    """
    results = []
    target_col_lower = target_column.lower()  # Make search case-insensitive
    
    # Determine which databases to scan
    if target_database:
        databases_to_search = [target_database]
    else:
        # Get all databases if no specific one is provided
        db_list = glue_client.get_databases()['DatabaseList']
        databases_to_search = [db['Name'] for db in db_list]
    
    for db_name in databases_to_search:
        try:
            # Get all tables in the current database
            tables_in_db = glue_client.get_tables(DatabaseName=db_name)['TableList']
            
            for table in tables_in_db:
                table_name = table['Name']
                # Extract the table's column schema from its storage descriptor
                table_columns = table.get('StorageDescriptor', {}).get('Columns', [])
                
                # Find columns that match the target (case-insensitive)
                matching_cols = [
                    col for col in table_columns 
                    if col['Name'].lower() == target_col_lower
                ]
                
                if matching_cols:
                    # Add the result with all relevant info
                    results.append({
                        'database': db_name,
                        'table': table_name,
                        'columns': matching_cols  # Includes name, data type, and comment (if exists)
                    })
                    
        except Exception as e:
            print(f"⚠️ Error accessing database {db_name}: {str(e)}")
            continue
    
    return results

How to Use the Function #

# Example 1: Search ALL databases for a column named 'customer_id'
matching_tables = find_tables_with_column('customer_id')

# Example 2: Search only the 'sales_db' database for 'order_date'
# matching_tables = find_tables_with_column('order_date', target_database='sales_db')

# Print results in a readable format
for item in matching_tables:
    print(f"\n📊 Database: {item['database']}")
    print(f"🔍 Table: {item['table']}")
    print("Matching Column Details:")
    for col in item['columns']:
        print(f"  - Name: {col['Name']}, Type: {col['Type']}")
        if 'Comment' in col:
            print(f"    Comment: {col['Comment']}")

Quick Notes #

• Case Sensitivity: The function uses case-insensitive matching (since Glue column names can sometimes have mixed cases). If you need exact case matching, remove the .lower() calls from both the target column and the column names in the schema.
• Permissions: Ensure your Glue development endpoint's IAM role has permissions for glue:GetDatabases and glue:GetTables actions.
• Large Catalogs: If you have hundreds of tables/databases, add pagination support (using NextToken in get_tables and get_databases calls) to handle large result sets smoothly.

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