データカタログとは

簡単にいうと，データ利活用を促進するために必要不可欠となる，データベースのメタ情報をまとめたもので，抽象的な概念である．そのため，何の情報があればデータカタログと言えるかはケースバイケースとなる．データカタログの必要性については，DMBOKで語られているため，この場では詳細については割愛する．

今回対象とするデータカタログの中身

大前提として，Snowflakeのデータベースのデータカタログを作成する．INFORMATION_SCHEMA で取得できる情報もあるが，そこから得られる情報は少なく，また，ダイナミックデータマスキングを用いると，ロールごとに見える情報も異なるため，Snowflakeでロール毎のお手製データカタログを作成することを考える．今回対象とするデータカタログの中身は以下の通りとなる．

ロール名，データベース名，スキーマ名，テーブル名，カラム名
レコード数
ユニークレコード数
Null レコード数
型
簡単なカラムの説明（テーブルやスキーマ，データベースの説明も作成できるが今回は割愛）

サンプルコード（修正中）

注意点

エラー処理は入れてません
最低限の動作を想定して，簡易なコードとなっています
事前に適切な権限が付与してください

import snowflake.snowpark as snowpark
from snowflake.snowpark.functions import col
import sys
import io
import pandas as pd

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
sys.stderr = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

def main(session: snowpark.Session): 
    session.use_warehouse("<WAREHOUSE_NAME>")    
    role_list = ["<ROLE_NAME_1>", "<ROLENAME_2>"]
    column_meta = pd.DataFrame(columns=["role","database_name","schema_name","table_name","column_name","count_r","count_u","count_null","dtype","metadata"])
    for x in range(0,len(role_list)):
        df_database_name = "<DATABASE_NAME>"
        df_role = role_list[x]
        session.use_database(df_database_name)
        session.use_role(df_role)
        df_schemas = session.sql("show terse schemas")
        df_schema = df_schemas.select(col('"name"'))
        pandas_df_schema = df_schema.to_pandas()
        for i in range(0,len(pandas_df_schema)):
            df_schema_name = pandas_df_schema.name[i]
            session.use_schema(df_schema_name)            
            pandas_df_table_name = session.sql("show terse views").select(col('"name"')).to_pandas()
            for j in range(0,len(pandas_df_table_name)):
                df_table_name = pandas_df_table_name.name[j]
                table_name_d = f"{df_database_name}.{df_schema_name}.{df_table_name}"
                df_table = session.table(table_name_d)
                df_table_columns = df_table.columns                
                count_r = df_table.count()
                for k in range(0,len(df_table_columns)):
                    df_table_columns_d = df_table_columns[k]
                    table_sql = f"select {df_table_columns_d},count(*) from {table_name_d} group by {df_table_columns_d} order by count(*)"
                    table_d = session.sql(table_sql)
                    count_u = table_d.count()
                    pd_table_d = table_d.select(col(df_table_columns_d)).to_pandas()                    
                    table_sqlnl = f"select count(CASE WHEN {df_table_columns_d} IS NULL THEN 1 END) AS NL from {table_name_d}"
                    table_null = session.sql(table_sqlnl)
                    table_null_pd = table_null.to_pandas()
                    count_null = table_null_pd.NL[0]
                    dtype = str(pd_table_d[df_table_columns_d].dtype)
                    meta_d = ""
                    meta_d = meta_d + f"カラム名は{df_table_columns_d}"
                    meta_d_sql = session.sql(f"select {df_table_columns_d} from {df_database_name}.{df_schema_name}.{df_table_name} limit 100")
                    meta_d_sql_pd = meta_d_sql.to_pandas()
                    for l in range(0,meta_d_sql_pd.shape[0]):
                        meta_d = meta_d + str(meta_d_sql_pd.iloc[j,0])
                        print(meta_d)
                    metadata_sql = f"select snowflake.cortex.complete('mistral-7b','{meta_d}に含まれるカラムの情報を端的に20文字以内で日本語で説明し、このとき{meta_d}に含まれる具体的なデータの中身は出力分に含めないようにしてください') as response;"
                    metadata = session.sql(metadata_sql)
                    metadata_pd = metadata.to_pandas()
                    column_meta.loc[len(column_meta)] = [df_role,df_database_name,df_schema_name,df_table_name,df_table_columns_d,count_r,count_u,count_null,dtype,metadata_pd.iloc[0][0]]
    snowflake_column_meta = session.create_dataframe(column_meta)
    snowflake_column_meta.write.mode("overwrite").save_as_table(f"{<T_DATABASE_NAME>}.{<T_SCHEMA_NAME>}.{<T_TABLE_NAME>}")
    return snowflake_column_meta