发布于: Nov 30, 2022
【概要】本文将接介绍一种数据库管理方法用于删除数据湖内的用户数据。通用数据保护条例（GDPR）是当今技术世界中的重要法规，也是众多在亚马逊云科技公有云当中建立解决方案的用户们所必须遵循的数据处理要求。
本文将接介绍一种数据库管理方法用于删除数据湖内的用户数据。通用数据保护条例（GDPR）是当今技术世界中的重要法规，也是众多在亚马逊云科技公有云当中建立解决方案的用户们所必须遵循的数据处理要求。GDPR 中提出一项“删除权”，或者叫“被遗忘权”条款，要求通过实施相关解决方案保证删除特定用户的个人数据。
在亚马逊云科技大数据与分析生态系统的背景之下，每一套架构，无论其面向怎样的目标，都需要使用 Amazon Simple Storage Service (Amazon S3) 作为核心存储服务。尽管 Amazon S3 拥有丰富的功能选项与完整性，但却缺少一种开箱即用的机制将用户标识符同包含用户数据的 S3 对象映射起来。
在本文中，我们将介绍一套框架，帮忙清除您组织中的亚马逊云科技托管数据湖内的各特定用户数据。此外，我们还将共同了解一套由多种不同亚马逊云科技存储层构成的分析解决方案，以及针对 Amazon S3 的示例代码。
为了解决数据清除框架实施中的种种挑战，我们在这里将问题简化为一个简单用例，即如何在使用亚马逊云科技作为数据管道的平台当中实现用户数据删除。下图说明了用例的基本情况。
我们引入了建立并维护索引元存储库的想法，该存储库能够跟踪每位用户的记录位置，帮助我们高效找出这些位置，从而缩小搜索空间。
您可以使用以下架构，在组织的亚马逊云科技数据湖内删除特定用户的数据。
对于此初始版本，我们创建了三个用户流，这些用户流负责将各项任务与合适的亚马逊云科技服务映射起来：
用户流 1：实时元数据存储更新
S3 ObjectCreated 或 ObjectDelete 事件会触发一项 Amazon Lambda 函数，此函数将解析对象并执行添加/更新/删除操作，以使元数据索引保持最新。您也可以为任意其他存储层建立类似的简单工作流，具体包括 Amazon Relational Database Service (RDS), Amazon Aurora 或 Amazon Elasticsearch Service (ES)。在本示例中，我们使用 Amazon DynamoDB 与 Amazon RDS for PostgreSQL 作为索引元数据存储选项，这里使用的具体方法广泛适用于其他技术场景。
用户流 2：清除数据
当用户要求删除其数据时，我们会通过 Amazon CloudWatch 触发一个 Amazon Step Functions 状态机来协调工作流。第一步是触发 Lambda 函数，由该函数查询元数据以识别出包含用户记录的存储层，并将生成的报告保存在 S3 报告存储桶内。接下来，由基于 Lambda Node JS 的工作节点创建并获取 Step Functions 活动，并通过 Amazon Simple Email Service (SES) 将带有批准及拒绝链接的电子邮件发送给审核人员。
下图所示为亚马逊云科技管理控制台上显示的 Step Functions 状态机基本架构。
审核人员从两条链接中选择其一，而后调用 Amazon API Gateway 端点，由该端点调用 Step Functions 以恢复工作流。如果选择批准链接，则 Step Functions 将触发一项 Lambda 函数，此函数将存储桶内的报告作为输入，据此删除存储层内的对象或记录，而后更新索引元存储库。在清除作业完成之后，Amazon Simple Notification Service (SNS) 会向用户发送操作成功或失败的通知邮件。
下图所示，为清除流程成功完成之后，控制台上的实际 Step Functions 执行流。
关于完整代码库，请参阅 GitHub repo 中的 step-function-definition.json 文件。
用户流 3：批量元数据存储更新
此用户流主要面向需要创建索引元存储的现有数据湖用例。您可以通过 Amazon Step Functions 进行流程编排，将历史数据作为输入并通过批处理作业更新元存储库。本文中的实现方案并不包含此用户流的示例脚本。
现在，我们将具体介绍实现过程中使用的两个用例：
- 您在每个 Amazon S3 文件中存储有多条用户记录
- 用户将各记录存储在同类 Amazon Web Service 存储层内
以这两种用例为基础，我们将演示实现索引元数据存储的几种替代方法。
按 S3 URI 与行号建立索引
在此用例中，我们可以使用免费的 RDS Postgres 实例实现索引存储。首先，使用以下代码创建一个简单表：
CREATE UNLOGGED TABLE IF NOT EXISTS user_objects ( userid TEXT, s3path TEXT, recordline INTEGER );
您可以按 user_id 建立索引，借此优化查询性能。在上传对象时，您需要在 user_objects 表中插入用于标识用户 ID 的行、标识目标 Amazon S3 对象的 URI，以及对应的行记录。例如，当上传以下 JSON 输入时，请输入以下代码：
{"user_id":"V34qejxNsCbcgD8C0HVk-Q","body":"…"}
{"user_id":"ofKDkJKXSKZXu5xJNGiiBQ","body":"…"}
{"user_id":"UgMW8bLE0QMJDCkQ1Ax5Mg","body ":"…"}
我们将 Amazon S3 位置 s3://gdpr-demo/year=2018/month=2/day=26/input.json 元组信息插入到 user_objects 表中，详见以下代码：
(“V34qejxNsCbcgD8C0HVk-Q”, “s3://gdpr-demo/year=2018/month=2/day=26/input.json”, 0) (“ofKDkJKXSKZXu5xJNGiiBQ”, “s3://gdpr-demo/year=2018/month=2/day=26/input.json”, 1) (“UgMW8bLE0QMJDCkQ1Ax5Mg”, “s3://gdpr-demo/year=2018/month=2/day=26/input.json”, 2)
您可以在任意 Amazon S3 ObjectCreated 事件上触发 Lambda 函数，借此实现对索引的更新操作。
当我们收到来自用户的删除请求时，则需要查询索引以获取关于数据存储位置的相关信息。具体请参见以下代码：
SELECT s3path,
ARRAY_AGG(recordline)
FROM user_objects
WHERE userid = ‘V34qejxNsCbcgD8C0HVk-Q’
GROUP BY;
以上示例 SQL 查询将返回如下行：(“s3://gdpr-review/year=2015/month=12/day=21/review-part-0.json“, {2102,529})
输出表明，S3 对象 s3://gdpr-review/year=2015/month=12/day=21/review-part-0.json 中的第 529 行与第 2102 行中包含请求的用户数据，需要清除。接下来，我们需要下载对象、删除这些行，然后覆盖对象。关于实现此功能的 Lambda 函数的 Python 实现，请参阅 GitHub repo 中的 deleteUserRecords.py。
可用记录行能够帮助我们以字节格式高效执行删除操作。为了简化实施过程，我们使用空的 JSON 对象替换已删除的行，借此快速实现行清除。此项操作只会带来少量存储开销，且消除了对索引内的后续元数据行进行更新的高成本操作需求。要消除空 JSON 对象，我们可以采用离线 vaccum 配合索引更新的方式。
按文件名索引，按索引键分组
在此用例中，我们创建了一个 DynamoDB 表以存储索引信息。之所以选择 DynamoDB，是因为其拥有良好的易用性与可扩展性；您可以使用按需计费模型，因此无需猜测可能需要的具体容量单位。在将文件上传至数据湖后，Lambda 函数将解析文件名（例如 1001-.csv）以标记用户标识符，并据此填充 DyanmoDB 元数据表。Userid为分区键，每个不同存储层都拥有自己的属性。例如，如果用户 1001 在 Amazon S3 及 Amazon RDS 中拥有数据，则其记录将类似于以下形式：
{"userid:": 1001, "s3":{"s3://path1", "s3://path2"}, "RDS":{"db1.table1.column1"}}
关于此功能的 Python 实例示例，请参阅 GitHub repo 中的 update-dynamo-metadata.py。
根据删除请求，我们需要查询元数据存储表（即 DynamoDB）并生成清除报告，该报告中包含关于那些存储层内包含用户记录的详细信息，同时提供有助于加快记录查找速度的其他提示信息。我们将清除报告存储在 Amazon S3 当中。关于实现此逻辑的示例 Lambda 函数，请参阅 GitHub repo 中的 generate-purge-report.py。
在清除获得批准之后，我们将使用清除报告作为输入，借此删除所有对应资源。关于 Lambda 函数的实现示例，请参阅 GitHub repo 中的 gdpr-purge-data.py。