QuickSIghtでどのデータセットの容量が大きいかを知る方法

QuickSightのSPICEデータベースの容量、
どのデータセットが多く使っているのかを知りたいのですが、
画面上ではそれを知る方法がありません。

pythonのAWSSDK(boto3)にQuickSightのAPIがあるので、これらを使えばできそうです。

• list_data_sets : データセット一覧取得
• describe_data_set : データセットの詳細取得（データサイズをここから取得）
• describe_data_set_permissions : データセットへの権限を取得（削除権限保有者を判別）

概要

Lambdaでコマンドをたたき、ログ上に容量の大きいデータセットTop10を表示するようにします。

ランタイムはPython3.8を使います。
タイムアウトは処理時間に応じて数分程度に伸ばしてセットします。
Lambdaのロールには上記のAPIを呼べるような権限をセットします。

pandasを使うので、Klayersを使ってレイヤにpandasを登録します。
参考：AWS LambdaでPython外部ライブラリのLayerを作る前に

Lambda

以下のようなLambdaを作成し、TESTなどで実行すると、ログにTop10が表示されます。

spice_checker

import logging
import re

import boto3
import pandas as pd
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

AWS_ACCOUNT_ID = '～AWSのアカウントID～'
client = boto3.client('quicksight')


def lambda_handler(event, context):

    # データセット一覧を取得
    df_data_set = list_data_sets()

    # データサイズ、所有者を追加
    df_data_set = describe_data_set(df_data_set)
    
    # 見栄えを整えてログ表示
    return show_data_sets(df_data_set)


# データセット一覧取得
def list_data_sets():
    df = pd.DataFrame(index = [], columns = ['Dataset_Id', 'Name', 'Created_Time', 'Last_Updated_Time', 'Import_Mode', 'Author', 'Spice_Bytes', 'Status'])
    
    response = client.list_data_sets(AwsAccountId = AWS_ACCOUNT_ID)
    data_sets = response['DataSetSummaries']
    
    for data_set in data_sets:
        dataset_id = data_set['DataSetId']
        name = data_set['Name']
        # 時間関連はUTCなのでJSTに変換してから取り込む
        created_time = pd.to_datetime(data_set['CreatedTime'], utc=True).tz_convert('Asia/Tokyo')
        last_updated_time = pd.to_datetime(data_set['LastUpdatedTime'], utc=True).tz_convert('Asia/Tokyo')
        import_mode = data_set['ImportMode']
        ds = pd.Series([dataset_id, name, created_time, last_updated_time, import_mode, '', 0, False], index=df.columns)
        df = df.append(ds, ignore_index=True)
    
    return df


# 指定されたデータセットの詳細情報（サイズ、所有者）を取得
def describe_data_set(df_data_set):
    for index, row in df_data_set.iterrows():
        spice_bytes = 0
        dataset_id = row['Dataset_Id']
        try:
            # 詳細情報取得
            response = client.describe_data_set(
                AwsAccountId = AWS_ACCOUNT_ID,
                DataSetId = dataset_id
            )
            ds = response['DataSet']

            df_data_set.at[index, 'Spice_Bytes'] = ds['ConsumedSpiceCapacityInBytes']
            df_data_set.at[index, 'Status'] = True  # サイズ取得成功かどうかの判別用

        except ClientError as e:
            # データセットによってはInvalidParameterValueExceptionが発生する
            if e.response['Error']['Code'] == 'InvalidParameterValueException':
                logger.info(row)
                logger.error(type(e)) 
            else:
                raise e
        
        # 所有者取得
        df_data_set.at[index, 'Author'] = get_author(dataset_id)
        
    return df_data_set


# 所有者を返す。複数の場合はカンマ区切り
def get_author(dataset_id):
    # 所有者の判定方法がわからなかったのでdelete権限を持っているユーザを返すようにしているので注意
    author_list = []
    response = client.describe_data_set_permissions(
        AwsAccountId = AWS_ACCOUNT_ID,
        DataSetId = dataset_id
    )
    permissions = response['Permissions']
    if len(permissions) == 0:
        loccer.info('権限保有者なし')
        return ""
    
    for permission in permissions:
        if 'quicksight:DeleteDataSet' in permission['Actions']:
            # ARN文字列からユーザ名を取り出す
            author = re.sub(r'arn:aws:quicksight:.*:.*:user\/.*\/(.*)', r'\1', permission['Principal'])
            author_list.append(author)
    
    return ",".join(author_list)


# サイズの大きいデータセット順にログ表示
def show_data_sets(df_data_set):
    # ログ出力の見た目を整える
    pd.set_option('display.max_columns', 100)
    pd.set_option("display.width", 200)
    pd.set_option('display.unicode.east_asian_width', True)
    df_data_set = df_data_set.sort_values('Spice_Bytes', ascending=False)
    
    # 表示用の列を追加
    view_column = ['Name', 'Created_YMD', 'Last_Update_YMD', 'Author', 'Spice_MBytes']
    df_data_set['Created_YMD'] = df_data_set['Created_Time'].dt.strftime('%Y/%m/%d')
    df_data_set['Last_Update_YMD'] = df_data_set['Last_Updated_Time'].dt.strftime('%Y/%m/%d')
    df_data_set['Spice_MBytes'] = (df_data_set['Spice_Bytes'] / 1024 / 1024).astype(int)

    # 結果をログに出力
    print("---------------------------------")
    print("サイズが大きいdata setのTop 10")
    print("---------------------------------")
    print(df_data_set[view_column].head(10))
    print("---------------------------------")
    
    # 将来的にはAPI的に使えるようにデータを返しておく
    return df_data_set.to_json(orient='records', force_ascii=False)

注意点

InvalidParameterValueException

describe_data_set のapiを呼んだ時にデータセットによっては InvalidParameterValueException のエラーが発生します。
どういうデータセットだとこのエラーになってしまうのかはよくわかっていません。
ご存知の方はお知らせいただければ幸いです。

このエラーが発生したときはexceptして0byteとして計上していますので、ご注意ください。

作成者の判定

データセットの作成者の判定方法がわかっていません。
よって describe_data_set_permissions のAPIで削除権限を持っている人を作成者として判定しています。
QuickSightの管理者画面では所有者を表示できているので
何らかの方法で取れそうなものなのですが、、、