BigQueryに作成されるテーブルは放置するとデータ保存必要がどんどん膨らむ。
そのため、テーブルの最終アクセス日を確認するクエリを作成。
プロジェクト横断して結果確認できるように組織権限が必要だったのと
データフレーム化のためにセッション権限が必要だったところがちょっと苦労した。

前提条件

to_dataframeを実行するときに、readsessions権限が必要。
次の1、2のいずれかで権限を付与しておく。

権限を個別指定する場合、次の3つを付与
- bigquery.readsessions.create
- bigquery.readsessions.getData
- bigquery.readsessions.update
下記基本ロールのいずれかを付与
- BigQuery 読み取りセッションユーザー
- BigQuery ユーザー
- BigQuery 管理者

また、プロジェクトを横断してINFORMATION_SCHEMAを参照するため
組織権限で次の権限を付与したロールを作成し実行ユーザーに付与してもらう。

bigquery.datasets.get
bigquery.jobs.list
bigquery.jobs.listAll
bigquery.reservationAssignments.list
bigquery.routines.get
bigquery.routines.list
bigquery.tables.get
bigquery.tables.getData
bigquery.tables.list

必要なモジュールをインストール

$ pip install pandas db-dtypes google-cloud-bigquery

ソースコード

実行すると、プロジェクト横断したテーブル/ビューの最終アクセス日が確認できる。
注意点としては、2022年4月20日以降にスキーマまたはデータが変更されていないテーブルは
対象行が存在しないため、当該テーブル/ビューはnullになるケースがある

なお、ローカルから実行するときはBigQueryにアクセスする認証情報を承認するため
下記のコマンドを実行し、認証情報を取得しておくと良い。

$ gcloud auth application-default login

main.py

from google.cloud import bigquery

class SQL:
    
    @staticmethod
    def get_project_ids():
        _SQL = """
        SELECT
          DISTINCT project_id
        FROM
          `region-asia-northeast1`.INFORMATION_SCHEMA.JOBS_BY_ORGANIZATION
        """
        return(_SQL)
    
    @staticmethod
    def get_information_schema(project_id_list:list):
       
        repeat_string_org = """
          SELECT
            table_catalog as project_id
            ,table_schema as dataset_id
            ,table_name as table_id
            ,table_type
            ,DATE(DATETIME_TRUNC(creation_time, DAY, '+9')) as created_date
          FROM
            `{}.region-asia-northeast1`.INFORMATION_SCHEMA.TABLES
        """
        repeat_string = ""
        id_length = len(project_id_list)
        if id_length == 1:
            repeat_string = repeat_string_org.format(project_id_list[0])
        elif id_length > 1:
            for i,project_id in enumerate(project_id_list):
                if i == 0:
                    repeat_string = repeat_string_org.format(project_id) + 'UNION DISTINCT'
                elif i == id_length - 1:
                    repeat_string += repeat_string_org.format(project_id)
                else:
                    repeat_string += repeat_string_org.format(project_id) + 'UNION DISTINCT'
        else:
            raise IndexError('project_id does not exist.' )

        _SQL = f"""
        WITH
        /*
        ネストされた配列[referenced_table]をフラット化
        */
          ORG AS
          (
          SELECT
            start_time
            ,referenced_table.project_id
            ,referenced_table.dataset_id
            ,referenced_table.table_id
          FROM
             `region-asia-northeast1`.INFORMATION_SCHEMA.JOBS_BY_ORGANIZATION
          CROSS JOIN
            UNNEST(referenced_tables) as referenced_table
        /*
        CTASやINSERTなどの対象テーブルを取得
        ドキュメントは配列とあるが、間違っている？
        */
        UNION DISTINCT
          SELECT
            start_time
            ,destination_table.project_id
            ,destination_table.dataset_id
            ,destination_table.table_id
          FROM
             `region-asia-northeast1`.INFORMATION_SCHEMA.JOBS_BY_ORGANIZATION
          )
        , TBL AS
        ( /* テーブルの作成日を取得 */
         {repeat_string}
        ),
        GET_MAX AS
        (
          SELECT
             project_id
            ,dataset_id
            ,table_id
            ,DATE(max(DATETIME_TRUNC(start_time, DAY, '+9'))) as referenced_date
          FROM
            ORG
          GROUP BY 1,2,3
        )
        
        SELECT
            TBL.project_id
            ,TBL.dataset_id
            ,TBL.table_id
            ,TBL.table_type
            ,TBL.created_date
            ,GET_MAX.referenced_date
            ,DATE_DIFF(GET_MAX.referenced_date,TBL.created_date, DAY) as diff_date --参照されてないときはnull
        FROM
          TBL
        LEFT JOIN
          GET_MAX
        USING
          (project_id,dataset_id,table_id)        
        """
        return(_SQL)

class BQ:

    def __init__(self):
        self.__client = bigquery.Client()

    @property
    def client(self):
        return(self.__client)

    def query_execute(self, SQL:str, get_dateframe:bool=False):
        """
        Get Query Result

        Parameters
        ----------
        SQL : string
        get_dataframe : bool. if True, you get DataFrame, else, you get RowIterator

        Returns
        -------
        google.cloud.bigquery.table.RowIterator
        """
        if not get_dateframe:
            rows = self.client.query(SQL).result()
        else:
            rows = self.client.query(SQL).to_dataframe()
        return(rows)

def main():
    bq = BQ()
    
    # Get Project ID List
    res = bq.query_execute(SQL.get_project_ids())
    row_list = []
    for row in res:
        row_list.extend([row.project_id])
    # print(row_list)    
    # Show SQL
    # print(SQL.get_information_schema(row_list))
    # Get Result
    df = bq.query_execute(SQL.get_information_schema(row_list), True)
    print(df) # デーフレーム化せずに、BigQuery内にテーブル作ってもOK
    
    
if __name__=='main':
    main()

参考

Pythonライブラリの使用
 BigQuery IAMによるアクセス制御
 GitHub googleapis/python-bigquery
Python Client for Google BigQuery
ぽ靴な缶