클래식 API와 함께 데이터 분류 사용하기¶

개요¶

데이터 분류에 대한 다음 예제에서는 범위를 기준으로 다양한 사용 사례를 다룹니다.

• 단일 테이블 분류하기

• 스키마의 모든 테이블 분류하기

• 데이터베이스의 모든 테이블 분류하기

새로운 분류 사용자라면 단일 테이블 분류로 시작한 다음 다른 두 예제로 진행하십시오.

공통 테이블 및 역할¶

이 항목에 나오는 예제에서는 아래에 표시된 테이블과 사용자 지정 역할을 사용합니다.

• 테이블: 직원에 대한 데이터를 포함한 my_db.my_schema.hr_data.

• 역할:

  • data_engineer: 테이블을 분류합니다.

  • policy_admin: 마스킹 정책으로 개인 정보를 보호합니다.

  • analyst: 액세스를 제한하고 싶은 사용자 지정 역할의 예로 사용됩니다. 예제에서는 이 역할이 존재한다고 가정합니다.

  • 데이터베이스 역할:

    • GOVERNANCE_ADMIN: 분류 시스템 태그를 할당합니다.

    • GOVERNANCE_VIEWER: 태그 및 마스킹 정책과 관련된 Account Usage 뷰를 쿼리합니다.

1. 테이블 만들기:

   CREATE OR REPLACE TABLE hr_data (
       age INTEGER,
       email_address VARCHAR,
       fname VARCHAR,
       lname VARCHAR
       );
   

   Copy

2. 테이블을 로딩합니다(세부 정보는 표시되지 않음).

3. 사용자 지정 역할을 만들고 그와 같은 역할에 필요한 권한을 부여합니다.

4. data_engineer 역할이 분류 프로세스를 실행하려면 테이블에 대한 액세스 권한이 필요합니다. 역할을 데이터베이스 역할로 전환할 수 없으므로 이 예제에서는 data_engineer 역할에 GOVERNANCE_ADMIN 데이터베이스 역할을 부여합니다.

   use role accountadmin;
   
   create role data_engineer;
   
   grant usage on database my_db to role data_engineer;
   
   grant usage on schema my_db.my_schema to role data_engineer;
   
   grant select, update on table my_db.my_schema.hr_data to role data_engineer;
   
   grant database role snowflake.governance_admin to role data_engineer;
   

   Copy

5. policy_admin 사용자 지정 역할은 PII 데이터를 포함하는 모든 열에 마스킹 정책을 할당해야 합니다.

   use role accountadmin;
   create role policy_admin;
   grant apply masking policy on account to role policy_admin;
   grant database role snowflake.governance_viewer to role policy_admin;
   

   Copy

단일 테이블 분류하기¶

단일 테이블 예제에서는 3단계 분류 프로세스(분석, 검토, 적용)를 확장하여 결과를 적용해 analyst 사용자 지정 역할에서 테이블 액세스 권한을 취소합니다.

USE ROLE SECURITYADMIN;

SELECT * FROM SNOWFLAKE.ACCOUNT_USAGE.TAG_REFERENCES
    WHERE TAG_NAME = 'PRIVACY_CATEGORY'
    AND TAG_VALUE= 'IDENTIFIER';

REVOKE SELECT ON TABLE my_db.my_schema.hr_data FROM ROLE analyst;

스키마의 모든 테이블 분류하기¶

이 예에서는 두 개의 사용자 정의 저장 프로시저를 사용하여 스키마의 모든 테이블을 분류하는 방법을 보여줍니다.

• classify_schema: 스키마의 모든 테이블을 나열하고, 분류 결과를 저장할 테이블을 만든 다음, 각 테이블에서 분류 태그를 추출하여 결과 테이블에 저장합니다.

• associate_tag_batch: classify_schema 저장 프로시저의 결과를 사용하여 분류 시스템 태그를 스키마의 모든 테이블 열에 자동으로 할당하고 각 테이블에 할당된 태그 수를 반환합니다.

1. classify_schema 라는 첫 번째 프로시저를 만듭니다.

   create or replace procedure classify_schema(schema_name string, result_table string)
   returns object language JavaScript
   as $$
   // 1
   const table_names = snowflake.execute({
     sqlText: `show terse tables in schema identifier(?)`,
     binds: [SCHEMA_NAME],
   });
   
   // helper function for quoted table names
   function quote(name) {
     return '"'+ name.replace(/"/g,'""') + '"';
   }
   
   // create table to store results in. if it already exists, we will add to it rather than overwrite
   snowflake.execute({
       sqlText: `create temp table if not exists identifier(?) (table_name string, result variant)`,
       binds: [RESULT_TABLE],
   })
   // loop through tables
   while (table_names.next()) {
     let name = table_names.getColumnValue('name');
     // add schema to table name
     name = SCHEMA_NAME + "." + quote(name);
     // insert qualified table name and result into result table
     const results = snowflake.execute({
       sqlText: `insert into identifier(?) select ?, extract_semantic_categories(?)`,
       binds: [RESULT_TABLE, name, name],
     });
   }
   // return the number of tables classified
   return {tables_classified: table_names.getRowCount()};
   $$;
   

   Copy

2. associate_tag_batch 라는 두 번째 프로시저를 만듭니다.

   create or replace procedure associate_tag_batch(result_table string)
   returns Object language JavaScript
   as $$
   // get table names and classification results to loop through
   const tags_to_apply = snowflake.execute({
     sqlText: `select table_name, result from identifier(?)`,
     binds: [RESULT_TABLE],
   });
   
   const out = {};
   while (tags_to_apply.next()) {
     // get table name
     const name = tags_to_apply.getColumnValue('TABLE_NAME');
     // get classification result
     const classification_results = tags_to_apply.getColumnValue('RESULT');
     // call associate semantic category tags with table name and classification result
     const results = snowflake.execute({
       sqlText: `call associate_semantic_category_tags(?, parse_json(?))`,
       binds: [name, JSON.stringify(classification_results)],
     });
     results.next();
     out[name] = results.getColumnValue(1).split('\n');
   }
   // return number of tags applied per table
   return out;
   $$;
   

   Copy

3. 분류하려는 스키마의 이름과 각 테이블에 대한 EXTRACT_SEMANTIC_CATEGORY의 결과를 가진 임시 테이블의 이름을 인자로 사용하여 classify_schema 저장 프로시저를 호출합니다.

   call classify_schema('my_db.my_schema','my_temporary_classification_table');
   

   Copy

4. 임시 테이블의 결과를 검토하고 필요에 따라 수정합니다.

5. 결과에 만족하면 다음과 같이 associate_tag_batch 저장 프로시저를 호출하여 분류 시스템 태그를 테이블 열에 할당합니다.

   call associate_tag_batch('my_temporary_classification_table');
   

   Copy

데이터베이스의 모든 테이블 분류하기¶

이 예에서는 두 개의 저장 프로시저를 사용하여 데이터베이스의 모든 테이블을 분류하는 방법을 보여줍니다.

• classify_database: 데이터베이스의 각 스키마에 대해 스키마 내의 모든 테이블을 분류하고 분류된 테이블과 스키마의 수를 반환합니다.

• associate_tag_batch: 이 항목의 스키마의 모든 테이블 분류하기 에 정의된 것과 동일한 작업을 수행합니다.

1. classify_database 저장 프로시저를 만듭니다.

   create or replace procedure classify_database(database_name string, result_table string)
   returns Object language JavaScript
   as $$
   // get list of schemas in database
   const schema_names = snowflake.execute({
     sqlText: `show terse schemas in database identifier(?)`,
     binds: [DATABASE_NAME],
   });
   
   // helper function for quoted schema names
   function quote(name) {
     return '"'+ name.replace(/"/g,'""') + '"';
   }
   
   // counter for tables. will use result from classify_schema to increment
   let table_count = 0
   while (schema_names.next()) {
     let name = schema_names.getColumnValue('name');
     // skip the information schema
     if (name == "INFORMATION_SCHEMA") {
       continue;
     }
     // add database name to schema
     name = DATABASE_NAME + "." + quote(name);
     // call classify_schema on each schema. This will loop over tables in schema
     const results = snowflake.execute({
       sqlText: `call classify_schema(?, ?)`,
       binds: [name, RESULT_TABLE],
     });
     results.next();
     // increment total number of tables by the number of tables in the schema
     table_count += results.getColumnValue(1).tables_classified ;
   }
   
   return {
       tables_classified: table_count,
       // subtract one from number of schemas because we skip the information schema
       schemas_classified: schema_names.getRowCount() - 1,
   };
   $$;
   

   Copy

2. 분류하려는 데이터베이스의 이름과 데이터베이스의 각 스키마에 결과를 저장할 임시 테이블의 이름으로 classify_database 저장 프로시저를 호출합니다.

   call classify_database('my_db','my_temporary_classification_table');
   

   Copy

3. 각 스키마로 이동하여 임시 테이블을 검토하고 필요한 경우 수정합니다.

4. 결과에 만족하면 다음과 같이 각 스키마에 대해 associate_tag_batch 저장 프로시저를 한 번 호출하여 해당 스키마의 테이블에 태그를 적용합니다. 예를 들어 데이터베이스에 세 개의 스키마가 포함된 경우 저장 프로시저를 세 번 호출합니다.

   call associate_tag_batch('my_temporary_classification_table');
   

   Copy