使用敏感数据保护和 Dataflow 自动将 Data Catalog 中的敏感度标记应用于文件、数据库和 BigQuery 表

设置您的环境

1. 在 Cloud Shell 中，克隆源代码库，然后前往克隆文件的目录：

   git clone https://github.com/GoogleCloudPlatform/auto-data-tokenize.git
   
   cd auto-data-tokenize/
   

2. 使用文本编辑器修改 set_variables.sh 脚本以设置所需的环境变量。忽略脚本中的其他变量。 它们与本文档无关。

   # The Google Cloud project to use:
   export PROJECT_ID="PROJECT_ID"
   
   # The Compute Engine region to use for running dataflow jobs and create a
   # temporary storage bucket:
   export REGION_ID= "REGION_ID"
   
   # The Cloud Storage bucket to use as a temporary bucket for Dataflow:
   export TEMP_GCS_BUCKET="CLOUD_STORAGE_BUCKET_NAME"
   
   # Name of the service account to use (not the email address)
   # (For example, tokenizing-runner):
   export DLP_RUNNER_SERVICE_ACCOUNT_NAME="DLP_RUNNER_SERVICE_ACCOUNT_NAME"
   
   # Entry Group ID to use for creating/searching for Entries
   # in Data Catalog for non-BigQuery entries.
   # The ID must begin with a letter or underscore, contain only English
   # letters, numbers and underscores, and have 64 characters or fewer.
   export
   DATA_CATALOG_ENTRY_GROUP_ID="DATA_CATALOG_ENTRY_GROUP_ID"
   
   # The Data Catalog Tag Template ID to use
   # for creating sensitivity tags in Data Catalog.
   # The ID must contain only lowercase letters (a-z), numbers (0-9), or
   # underscores (_), and must start with a letter or underscore.
   # The maximum size is 64 bytes when encoded in UTF-8
   export INSPECTION_TAG_TEMPLATE_ID="INSPECTION_TAG_TEMPLATE_ID"
   

   请替换以下内容：

   • PROJECT_ID：您的项目 ID。
   • REGION_ID：包含一个或多个存储桶的区域。选择 Data Catalog 区域中的位置。
   • CLOUD_STORAGE_BUCKET_NAME：存储桶的名称。
   • DLP_RUNNER_SERVICE_ACCOUNT_NAME：您的服务账号的名称。
   • DATA_CATALOG_ENTRY_GROUP_ID：非 BigQuery 数据目录条目组的名称。
   • INSPECTION_TAG_TEMPLATE_ID：您为 Data Catalog 标记模板指定的名称

3. 运行该脚本以设置环境变量：

   source set_variables.sh
   

创建资源

本文档中介绍的架构使用以下资源：

• 用于运行 Dataflow 流水线的服务账号，以便启用精细的访问权限控制
• 用于存储临时数据和测试数据的 Cloud Storage 存储桶
• 用于将敏感度标记附加到条目的 Data Catalog 标记模板
• 作为 JDBC 来源的 MySQL on Cloud SQL 实例

创建服务账号

我们建议您运行具有精细访问权限控制的流水线以改进访问分区。如果您的项目没有用户创建的服务账号，请创建一个。

1. 在 Cloud Shell 中，创建一个服务账号来用作 Dataflow 的用户管理控制器的服务账号：

     gcloud iam service-accounts create ${DLP_RUNNER_SERVICE_ACCOUNT_NAME} \
     --project="${PROJECT_ID}" \
     --description="Service Account for Sampling and Cataloging pipelines." \
     --display-name="Sampling and Cataloging pipelines"
   

2. 创建具有访问敏感数据保护、Dataflow、Cloud SQL 和 Data Catalog 所需权限的自定义角色：

     export SAMPLING_CATALOGING_ROLE_NAME="sampling_cataloging_runner"
     gcloud iam roles create ${SAMPLING_CATALOGING_ROLE_NAME} \
     --project="${PROJECT_ID}" \
     --file=tokenizing_runner_permissions.yaml
   

3. 将自定义角色和 Dataflow Worker 角色应用于该服务账号，使其以 Dataflow 工作器身份运行：

     gcloud projects add-iam-policy-binding ${PROJECT_ID} \
     --member="serviceAccount:${DLP_RUNNER_SERVICE_ACCOUNT_EMAIL}" \
     --role=projects/${PROJECT_ID}/roles/${SAMPLING_CATALOGING_ROLE_NAME}
   
     gcloud projects add-iam-policy-binding ${PROJECT_ID} \
     --member="serviceAccount:${DLP_RUNNER_SERVICE_ACCOUNT_EMAIL}" \
     --role=roles/dataflow.worker
   

创建 Cloud Storage 存储分区

• 在 Cloud Shell 中，创建一个 Cloud Storage 存储桶以存储测试数据和作为 Dataflow 暂存位置：

  gsutil mb -p ${PROJECT_ID} -l ${REGION_ID} "gs://${TEMP_GCS_BUCKET}"
  

创建 Data Catalog 条目组

Data Catalog 维护一个代表 Google Cloud 资源或其他资源的条目列表。条目按条目组进行组织。BigQuery 存在一个隐式条目组 (@bigquery)。您必须为其他类型的资源创建条目组。如需详细了解 Data Catalog 条目，请参阅使用文件集条目显示 Cloud Storage 中的文件。

在 Data Catalog 中，条目组就像包含条目的文件夹。条目表示数据资源。

• 在 Cloud Shell 中，创建一个新的条目组，流水线可以在其中为您的 MySQL 表添加条目：

  gcloud data-catalog entry-groups create \
  "${DATA_CATALOG_ENTRY_GROUP_ID}" \
  --project="${PROJECT_ID}" \
  --location="${REGION_ID}"
  

创建检查标记模板

• 在 Cloud Shell 中，创建 Data Catalog 标记模板以支持对带有敏感信息的敏感数据保护条目进行标记：

  gcloud data-catalog tag-templates create ${INSPECTION_TAG_TEMPLATE_ID} \
  --project="${PROJECT_ID}"  \
  --location="${REGION_ID}" \
  --display-name="Auto DLP sensitive categories" \
  --field=id=infoTypes,type=string,display-name="DLP infoTypes",required=TRUE \
  --field=id=inspectTimestamp,type=timestamp,display-name="Inspection run timestamp",required=TRUE
  

在 BigQuery 中创建检查结果表

• 在 Cloud Shell 中，创建 BigQuery 表以存储来自敏感数据保护的汇总发现结果：

  bq mk --dataset \
  --location="${REGION_ID}" \
  --project_id="${PROJECT_ID}"  \
  inspection_results
  
  bq mk --table \
  --project_id="${PROJECT_ID}"  \
  inspection_results.SensitivityInspectionResults \
  inspection_results_bigquery_schema.json
  

在 Cloud SQL 实例上设置 MySQL

对于数据源，请使用 Cloud SQL 实例。

1. 在 Cloud Shell 中，实例化 MySQL on Cloud SQL 实例并使用示例数据加载该实例：

   export SQL_INSTANCE="mysql-autodlp-instance"
   export SQL_ROOT_PASSWORD="root1234"
   
   gcloud sql instances create "${SQL_INSTANCE}" \
   --project="${PROJECT_ID}"  \
   --region="${REGION_ID}" \
   --database-version=MYSQL_5_7 \
   --root-password="${SQL_ROOT_PASSWORD}"
   

2. 将数据库密码保存在 Secret Manager 中。

   不应存储或记录数据库密码和其他 Secret 信息。Secret Manager 可让您安全地存储和检索此类 Secret。

   将 MySQL 数据库根密码存储为云 Secret：

   export SQL_PASSWORD_SECRET_NAME="mysql-password"
   
   printf $SQL_ROOT_PASSWORD |
   gcloud secrets create "${SQL_PASSWORD_SECRET_NAME}" \
   --data-file=- \
   --locations="${REGION_ID}" \
   --replication-policy="user-managed" \
   --project="${PROJECT_ID}"
   

将测试数据复制到 Cloud SQL 实例

测试数据是一个演示数据集，其中包含 5,000 个随机生成的名字和姓氏以及美国格式的电话号码。demonstration-dataset 表包含四列：row_id、person_name、contact_type、contact_number。您还可以使用自己的数据集。如果您使用自己的数据集，请记得调整本文档的在 BigQuery 中验证的建议值。要将随附的演示数据集 (contacts5k.sql.gz) 复制到您的 Cloud SQL 实例，请执行以下操作：

1. 在 Cloud Shell 中，将示例数据集复制到 Cloud Storage 以暂存到 Cloud SQL 中：

   gsutil cp contacts5k.sql.gz gs://${TEMP_GCS_BUCKET}
   

2. 在 Cloud SQL 实例中创建新的数据库：

   export DATABASE_ID="auto_dlp_test"
   
   gcloud sql databases create "${DATABASE_ID}" \
   --project="${PROJECT_ID}"  \
   --instance="${SQL_INSTANCE}"
   

3. 将 Storage Object Admin 角色授予您的 Cloud SQL 服务账号，以便它可以访问存储空间：

   export SQL_SERVICE_ACCOUNT=$(gcloud sql instances describe
   "${SQL_INSTANCE}" --project="${PROJECT_ID}" | grep
   serviceAccountEmailAddress: | sed "s/serviceAccountEmailAddress: //g")
   
   gsutil iam ch "serviceAccount:${SQL_SERVICE_ACCOUNT}:objectAdmin" \
   gs://${TEMP_GCS_BUCKET}
   

4. 将数据加载到新表中：

   gcloud sql import sql "${SQL_INSTANCE}" \
   "gs://${TEMP_GCS_BUCKET}/contacts5k.sql.gz" \
   --project="${PROJECT_ID}"  \
   --database="${DATABASE_ID}"
   

   如需详细了解如何将数据导入 Cloud SQL，请参阅导入和导出数据的最佳实践。

编译模块

• 在 Cloud Shell 中，编译模块以构建用于部署采样和识别流水线和标记化流水线的可执行文件：

   ./gradlew clean buildNeeded shadowJar -x test
  

  （可选）如需运行单元测试和集成测试，请移除 -x test 标志。如果尚未安装 libncurses5，请使用 sudo apt-get install libncurses5 在 Cloud Shell 中进行安装。

运行采样和识别流水线

采样和敏感数据保护识别流水线按以下顺序执行以下任务：

1. 从提供的来源中提取记录。例如，敏感数据保护标识方法仅支持扁平表，因此流水线会将 Avro、Parquet 或 BigQuery 记录展平，因为这些记录可以包含嵌套和重复字段。
2. 对所需样本的各个列进行采样，不包括 null 或空值。
3. 使用敏感数据保护识别敏感的 infoTypes 数据，方法是将样本批量处理为敏感数据保护可接受的批次大小（小于 500 Kb 并且少于 50,000 个值）。
4. 将报告写入 Cloud Storage 和 BigQuery 以供日后参考。
5. 在您提供标记模板和条目组信息时，创建 Data Catalog 实体。提供此信息时，流水线会根据相应的列针对 Data Catalog 中的条目创建敏感度标记。

创建 Dataflow Flex 模板

借助 Dataflow Flex 模板，您可以使用 Google Cloud 控制台、Google Cloud CLI 或 REST API 调用在 Google Cloud 上设置和运行流水线。本文档提供了 Google Cloud 控制台的说明。经典模板作为执行图暂存在 Cloud Storage 上，而 Flex 模板将流水线作为容器映像打包到项目的 Container Registry 中。通过 Flex 模板，您可以分离构建和运行流水线，并与计划流水线运行的编排系统集成。如需详细了解 Dataflow Flex 模板，请参阅评估要使用的模板类型。

Dataflow Flex 模板会将构建和暂存步骤与正在运行的步骤分开。为此，可以使用 DataflowStartFlexTemplateOperator 模块从 API 调用和 Cloud Composer 启动 Dataflow 流水线。

1. 在 Cloud Shell 中，定义用于存储模板规范文件的位置，该文件包含运行 Dataflow 作业所需的信息：

   export
   FLEX_TEMPLATE_PATH="gs://${TEMP_GCS_BUCKET}/dataflow/templates/sample-inspect-tag-pipeline.json"
   
   export
   FLEX_TEMPLATE_IMAGE="us.gcr.io/${PROJECT_ID}/dataflow/sample-inspect-tag-pipeline:latest"
   

2. 构建 Dataflow Flex 模板：

   gcloud dataflow flex-template build "${FLEX_TEMPLATE_PATH}" \
   --image-gcr-path="${FLEX_TEMPLATE_IMAGE}" \
   --service-account-email="${DLP_RUNNER_SERVICE_ACCOUNT_EMAIL}" \
   --sdk-language="JAVA" \
   --flex-template-base-image=JAVA11 \
   --metadata-file="sample_identify_tag_pipeline_metadata.json" \
   --jar="build/libs/auto-data-tokenize-all.jar" \
   --env="FLEX_TEMPLATE_JAVA_MAIN_CLASS=\"com.google.cloud.solutions.autotokenize.pipeline.DlpInspectionPipeline\""
   

运行流水线

采样和识别流水线会提取由 sampleSize 值指定的记录数。然后，它会展平每条记录，并使用敏感数据保护识别 infoTypes 字段（以识别敏感信息类型）。系统会计算 infoTypes 值，然后按列名称和 infoType 字段进行聚合，以构建敏感度报告。

• 在 Cloud Shell 中，启动采样和识别流水线，以识别数据源中的敏感列：

  export
  CLOUD_SQL_JDBC_CONNECTION_URL="jdbc:mysql:///${DATABASE_ID}?cloudSqlInstance=${PROJECT_ID}%3A${REGION_ID}%3A${SQL_INSTANCE}&socketFactory=com.google.cloud.sql.mysql.SocketFactory"
  
  gcloud dataflow flex-template run "sample-inspect-tag-`date +%Y%m%d-%H%M%S`" \
    --template-file-gcs-location "${FLEX_TEMPLATE_PATH}" \
    --region "${REGION_ID}" \
    --service-account-email "${DLP_RUNNER_SERVICE_ACCOUNT_EMAIL}" \
    --staging-location "gs://${TEMP_GCS_BUCKET}/staging" \
    --worker-machine-type "n1-standard-1" \
    --parameters sampleSize=2000 \
    --parameters sourceType="JDBC_TABLE" \
    --parameters inputPattern="Contacts" \
    --parameters reportLocation="gs://${TEMP_GCS_BUCKET}/auto_dlp_report/" \
    --parameters reportBigQueryTable="${PROJECT_ID}:inspection_results.SensitivityInspectionResults" \
    --parameters jdbcConnectionUrl="${CLOUD_SQL_JDBC_CONNECTION_URL}" \
    --parameters jdbcDriverClass="com.mysql.cj.jdbc.Driver" \
    --parameters jdbcUserName="root" \
    --parameters jdbcPasswordSecretsKey="projects/${PROJECT_ID}/secrets/${SQL_PASSWORD_SECRET_NAME}/versions/1" \
    --parameters ^:^jdbcFilterClause="ROUND(RAND() * 10) IN (1,3)" \
    --parameters dataCatalogEntryGroupId="projects/${PROJECT_ID}/locations/${REGION_ID}/entryGroups/${DATA_CATALOG_ENTRY_GROUP_ID}" \
    --parameters dataCatalogInspectionTagTemplateId="projects/${PROJECT_ID}/locations/${REGION_ID}/tagTemplates/${INSPECTION_TAG_TEMPLATE_ID}"
  

jdbcConnectionUrl 参数指定包含用户名和密码详细信息的 JDBC 数据库连接网址。与构建确切的连接网址有关的详细信息取决于您的数据库供应商和托管合作伙伴。如需详细了解如何连接到基于 Cloud SQL 的关系型数据库，请参阅使用 Cloud SQL 连接器进行连接。

该流水线会构造一个查询（如 SELECT * FROM [TableName]）来读取表记录以进行检查。

此查询可能会在数据库以及流水线中加载数据，尤其是对于大型表。您可以视情况优化要在数据库端检查的记录样本。为此，请插入 jdbcFilterClause 作为在本文档后面“在 BigQuery 中验证”部分中提供的代码示例中出现的查询的 WHERE 子句。

如需运行报告，您可以选择以下一个或多个报告接收器：

• reportLocation，用于将报告存储在 Cloud Storage 存储桶中
• report BIGQUERY_TABLE，用于将报告存储在 BigQueryTable 中
• dataCatalogEntryGroupId，用于在 Data Catalog 中创建和标记条目（如果 sourceType 为 BIGQUERY_TABLE，请省略此参数）

该流水线支持以下源类型。要确定 sourceType 和 inputPattern 参数的正确组合，请使用下表中列出的选项。

在这种情况下，您只使用 JDBC_TABLE 表。

流水线会检测敏感数据保护支持的所有标准 infoTypes。您可以使用 --observableinfoTypes 参数提供其他自定义 infoTypes。

采样和识别流水线有向无环图 (DAG)

下图显示了 Dataflow 执行 DAG。DAG 有两个分支。两个分支都从 ReadJdbcTable 开始，到 ExtractReport 结束。接着，生成报告或存储数据。

检索报告

采样和识别流水线输出以下文件：

• 源的 Avro 架构文件（或转换为 Avro 的架构）
• 每个敏感列对应一个文件，其中包含 infoType 信息和计数

如需检索报告，请执行以下操作：

1. 在 Cloud Shell 中，检索报告：

   mkdir -p auto_dlp_report/
   gsutil -m cp
   "gs://${TEMP_GCS_BUCKET}/auto_dlp_report/*.json"
   auto_dlp_report/
   

2. 列出所有已识别的列名称：

   cat auto_dlp_report/col-*.json | jq .columnName
   

   输出如下所示：

   "$.topLevelRecord.contact_number" "$.topLevelRecord.person_name"

3. 使用文件的 cat 命令查看已识别列的详细信息：

   cat auto_dlp_report/col-topLevelRecord-contact_number-00000-of-00001.json
   

   以下为 cc 列的代码段：

   { "columnName": "$.topLevelRecord.contact_number", "infoTypes": [{ "infoType": "PHONE_NUMBER", "count": "990" }] }

   • columnName 值异常，因为隐式将数据库行转换为 Avro 记录。
   • count 值因执行期间随机选择的样本而异。

验证 Data Catalog 中的敏感度标记

采样和识别流水线会创建一个新条目，并将敏感度标记应用于相应的列。

1. 在 Cloud Shell 中，检索为“联系人”表创建的条目：

   gcloud data-catalog entries describe Contacts \
     --entry-group=${DATA_CATALOG_ENTRY_GROUP_ID} \
     --project="${PROJECT_ID}"  \
     --location="${REGION_ID}"
   

   此命令会显示该表的详细信息，包括其架构。

2. 显示附加到此条目的所有敏感度标记：

   gcloud data-catalog tags list --entry=Contacts
     --entry-group=${DATA_CATALOG_ENTRY_GROUP_ID} \
     --project="${PROJECT_ID}"  \
     --location="${REGION_ID}"
   

3. 验证以下列中是否存在敏感度标记：contact_number、person_name。

   敏感数据保护识别的 infoTypes 数据可能包含一些误报类型。例如，它可以将 person_name 类型标识为 DATE 类型，因为某些随机 person_names 字符串可以是 4 月、5 月、6 月或其他日期。

   输出的敏感度标记详细信息如下：

   column: contact_number
   fields:
     infoTypes:
       displayName: DLP infoTypes
       stringValue: '[PHONE_NUMBER]'
     inspectTimestamp:
       displayName: Inspection run timestamp
       timestampValue: '2021-05-20T16:34:29.596Z'
   name: projects/auto-dlp/locations/asia-southeast1/entryGroups/sql_databases/entries/Contacts/tags/CbS0CtGSpZyJ
   template: projects/auto-dlp/locations/asia-southeast1/tagTemplates/auto_dlp_inspection
   templateDisplayName: Auto DLP sensitive categories
   ---
   column: person_name
   fields:
     infoTypes:
       displayName: DLP infoTypes
       stringValue: '[DATE, PERSON_NAME]'
     inspectTimestamp:
       displayName: Inspection run timestamp
       timestampValue: '2021-05-20T16:34:29.594Z'
   name: projects/auto-dlp/locations/asia-southeast1/entryGroups/sql_databases/entries/Contacts/tags/Cds1aiO8R0pT
   template: projects/auto-dlp/locations/asia-southeast1/tagTemplates/auto_dlp_inspection
   templateDisplayName: Auto DLP sensitive categories
   

在 BigQuery 中验证

Dataflow 流水线将汇总发现结果附加到提供的 BigQuery 表中。该查询会输出从 BigQuery 表中检索到的检查结果。

• 在 Cloud Shell 中，检查结果：

  bq query \
    --location="${REGION_ID}"  \
    --project="${PROJECT_ID}"  \
    --use_legacy_sql=false \
  'SELECT
     input_pattern AS table_name,
     ColumnReport.column_name AS column_name,
     ColumnReport.info_types AS info_types
   FROM
     `inspection_results.SensitivityInspectionResults`,
     UNNEST(column_report) ColumnReport;
   WHERE column_name="$.topLevelRecord.contact_number"'
  

  输出如下所示：

  +------------+---------------------------------+----------------------------------------------+
  | table_name |           column_name           |                  info_types                  |
  +------------+---------------------------------+----------------------------------------------+
  | Contacts   | $.topLevelRecord.contact_number | [{"info_type":"PHONE_NUMBER","count":"990"}] |
  +------------+---------------------------------+----------------------------------------------+