创建将在不同环境中提取有关数据集和表大小的元数据信息的视图

Posted 2023-03-24

【中文标题】创建将在不同环境中提取有关数据集和表大小的元数据信息的视图

【英文标题】：Create View that will extract metadata information about dataset and table sizes in different environments

【发布时间】：2019-10-15 13:12:55

【问题描述】：

我们需要监控不同环境中的表大小。 使用 Google 元数据 API 获取给定项目/环境的信息。 需要创建一个视图，它将提供 1.所有数据集是什么 2.每个数据集中有哪些表 3. 桌子尺寸 4. 数据集大小

【参考方案1】：

BigQuery 已经为您内置了这样的视图：INFORMATION_SCHEMA 是一系列视图，可以访问有关数据集、表和视图的元数据

例如，下面返回默认项目中所有数据集的元数据

SELECT * FROM INFORMATION_SCHEMA.SCHEMATA   

或 对于我的项目

SELECT * FROM my_project.INFORMATION_SCHEMA.SCHEMATA    

还有其他类似的表格视图

此外，还有一个元表可用于获取有关给定数据集中表的更多信息：__TABLES__SUMMARY 和 __TABLES__

SELECT * FROM `project.dataset.__TABLES__`

例如：

SELECT table_id,
    DATE(TIMESTAMP_MILLIS(creation_time)) AS creation_date,
    DATE(TIMESTAMP_MILLIS(last_modified_time)) AS last_modified_date,
    row_count,
    size_bytes,
    CASE
        WHEN type = 1 THEN 'table'
        WHEN type = 2 THEN 'view'
        WHEN type = 3 THEN 'external'
        ELSE '?'
    END AS type,
    TIMESTAMP_MILLIS(creation_time) AS creation_time,
    TIMESTAMP_MILLIS(last_modified_time) AS last_modified_time,
    dataset_id,
    project_id
FROM `project.dataset.__TABLES__`  

【讨论】：

所以在上面的 SQL 中，如果我必须查找项目中所有数据集的信息，那么我必须手动提及数据集名称，并且必须全部联合，或者有没有更好的方法，因为如果创建了一个新数据集，那么我必须再次手动添加它

如果您使用__TABLES__，则正确。但是您可以使用没有这种“限制”的INFORMATION_SCHEMA 视图 - 请参阅我的答案

【参考方案2】：

为了使自动化查询，以检查项目中的每个数据集，而不是与UNION ALL手动添加，您可以遵循@ZinkyZinky here给出的建议，并创建生成每次UNION ALL电话查询数据集.__ TABLES_。我没有设法在BigQuery中完全自动使用该解决方案，因为我没有找到一个方法来执行（这就是string_agg创造）作为字符串生成的命令。总之，我已成功地开发在Python液，加入在接下来的查询生成的字符串。你可以找到下面的代码。它还创建一个新的表，并将结果存储有：

from google.cloud import bigquery

client = bigquery.Client()

project_id = "wave27-sellbytel-bobeda"

# Construct a full Dataset object to send to the API.
dataset_id = "project_info"
dataset = bigquery.Dataset(".".join([project_id, dataset_id]))
dataset.location = "US"

# Send the dataset to the API for creation.
# Raises google.api_core.exceptions.Conflict if the Dataset already
# exists within the project.
dataset = client.create_dataset(dataset)  # API request
print("Created dataset .".format(client.project, dataset.dataset_id))

schema = [
    bigquery.SchemaField("dataset_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("table_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("size_bytes", "INTEGER", mode="REQUIRED"),
]

table_id = "table_info"

table = bigquery.Table(".".join([project_id, dataset_id, table_id]), schema=schema)
table = client.create_table(table)  # API request
print(
    "Created table ..".format(table.project, table.dataset_id, table.table_id)
)

job_config = bigquery.QueryJobConfig()
# Set the destination table
table_ref = client.dataset(dataset_id).table(table_id)
job_config.destination = table_ref

# QUERIES
# 1. Creating the UNION ALL list with the table information of each dataset
query = (
    r"SELECT string_agg(concat('SELECT * from `', schema_name, '.__TABLES__` '), 'union all \n') "
    r"from INFORMATION_SCHEMA.SCHEMATA"
)

query_job = client.query(query, location="US")  # API request - starts the query
select_tables_from_all_datasets = ""
for row in query_job:
    select_tables_from_all_datasets += row[0]

# 2. Using the before mentioned list to create a table.
query = (
    "WITH ALL__TABLES__ AS ()"
    "SELECT dataset_id, table_id, size_bytes FROM ALL__TABLES__;".format(select_tables_from_all_datasets)
)

query_job = client.query(query, location="US", job_config=job_config)  # job_config configures in which table the results will be stored.
for row in query_job:
    print row

print('Query results loaded to table '.format(table_ref.path))