通用数据保护条例的监管下，你的数据湖“断舍离”了吗？

2021 年 12 月 23 日
本文字数：6077 字
阅读完需：约 20 分钟

通用数据保护条例(GDPR)是当今技术世界中的重要法规，也是众多在亚马逊云科技公有云当中建立解决方案的用户们所必须遵循的数据处理要求。GDPR 中提出一项“删除权”，或者叫“被遗忘权”条款，要求通过实施相关解决方案保证删除特定用户的个人数据。

在亚马逊云科技大数据与分析生态系统的背景之下，每一套架构，无论其面向怎样的目标，都需要使用 Amazon Simple Storage Service(Amazon S3)作为核心存储服务。尽管 Amazon S3 拥有丰富的功能选项与完整性，但却缺少一种开箱即用的机制将用户标识符同包含用户数据的 Amazon S3 对象映射起来。

在本文中，我们将介绍一套框架，帮忙清除您组织中的亚马逊云科技托管数据湖内的各特定用户数据。此外，我们还将共同了解一套由多种不同亚马逊云科技存储层构成的分析解决方案，以及针对 Amazon S3 的示例代码。

Amazon Simple Storage Service

https://aws.amazon.com/cn/s3/

📢 想要了解更多数据湖仓的相关知识？敬请关注亚马逊云科技中国峰会北京站的大数据与智能湖仓分论坛！更多最新技术发布和实践创新的内容尽在 8.19-20 北京、9.15 深圳分会，快来点击图片报名吧～

参考架构

为了解决数据清除框架实施中的种种挑战，我们在这里将问题简化为一个简单用例，即如何在使用亚马逊云科技作为数据管道的平台当中实现用户数据删除。下图说明了用例的基本情况。

我们引入了建立并维护索引元存储库的想法，该存储库能够跟踪每位用户的记录位置，帮助我们高效找出这些位置，从而缩小搜索空间。

您可以使用以下架构，在组织的亚马逊云科技数据湖内删除特定用户的数据。

对于此初始版本，我们创建了三个用户流，这些用户流负责将各项任务与合适的亚马逊云科技服务映射起来：

用户流 1：实时元数据存储更新

Amazon S3 ObjectCreated 或 ObjectDelete 事件会触发一项 Amazon Lambda 函数，此函数将解析对象并执行添加/更新/删除操作，以使元数据索引保持最新。您也可以为任意其他存储层建立类似的简单工作流，具体包括 Amazon Relational Database Service(RDS),Amazon Aurora 或 Amazon Elasticsearch Service(ES)。在本示例中，我们使用 Amazon DynamoDB 与 Amazon RDS for PostgreSQL 作为索引元数据存储选项，这里使用的具体方法广泛适用于其他技术场景。

Amazon Relational Database Service

https://aws.amazon.com/cn/rds/

Amazon Aurora

https://aws.amazon.com/cn/rds/aurora/?aurora-whats-new.sort-by=item.additionalFields.postDateTime&aurora-whats-new.sort-order=desc

Amazon Elasticsearch Service

https://aws.amazon.com/cn/elasticsearch-service/

Amazon DynamoDB

https://aws.amazon.com/cn/dynamodb/

Amazon RDS for PostgreSQL

https://aws.amazon.com/cn/rds/postgresql/

用户流 2：清除数据

当用户要求删除其数据时，我们会通过 Amazon CloudWatch 触发一个 Amazon Step Functions 状态机来协调工作流。第一步是触发 Lambda 函数，由该函数查询元数据以识别出包含用户记录的存储层，并将生成的报告保存在 Amazon S3 报告存储桶内。接下来，由基于 Lambda Node JS 的工作节点创建并获取 Amazon Step Functions 活动，并通过 Amazon Simple Email Service(SES)将带有批准及拒绝链接的电子邮件发送给审核人员。

Amazon Simple Email Service

https://aws.amazon.com/cn/ses/

下图所示为亚马逊云科技管理控制台上显示的 Amazon Step Functions 状态机基本架构。

审核人员从两条链接中选择其一，而后调用 Amazon API Gateway 端点，由该端点调用 Amazon Step Functions 以恢复工作流。如果选择批准链接，则 Amazon Step Functions 将触发一项 Lambda 函数，此函数将存储桶内的报告作为输入，据此删除存储层内的对象或记录，而后更新索引元存储库。在清除作业完成之后，Amazon Simple Notification Service(SNS)会向用户发送操作成功或失败的通知邮件。

Amazon Simple Notification Service

https://aws.amazon.com/cn/sns/?whats-new-cards.sort-by=item.additionalFields.postDateTime&whats-new-cards.sort-order=desc

下图所示，为清除流程成功完成之后，控制台上的实际 Step Functions 执行流。

关于完整代码库，请参阅 GitHub repo 中的 step-function-definition.json 文件。

step-function-definition.json

https://github.com/aws-samples/data-purging-aws-data-lake/blob/master/Scripts/index-by-file-name/step-function-definition.json

用户流 3：批量元数据存储更新

此用户流主要面向需要创建索引元存储的现有数据湖用例。您可以通过 Amazon Step Functions 进行流程编排，将历史数据作为输入并通过批处理作业更新元存储库。本文中的实现方案并不包含此用户流的示例脚本。

我们的框架

现在，我们将具体介绍实现过程中使用的两个用例：

您在每个 Amazon S3 文件中存储有多条用户记录
用户将各记录存储在同类 Amazon Web Service 存储层内

以这两种用例为基础，我们将演示实现索引元数据存储的几种替代方法。

按 Amazon S3 URI 与行号建立索引

在此用例中，我们可以使用免费的 Amazon 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})

复制代码

输出表明，Amazon S3 对象s3://gdpr-review/year=2015/month=12/day=21/review-part-0.json中的第 529 行与第 2102 行中包含请求的用户数据，需要清除。接下来，我们需要下载对象、删除这些行，然后覆盖对象。关于实现此功能的 Lambda 函数的 Python 实现，请参阅 GitHub repo 中的 deleteUserRecords.py。

deleteUserRecords.py

https://github.com/aws-samples/data-purging-aws-data-lake/blob/master/Scripts/index-by-row-number/deleteUserRecords.py

可用记录行能够帮助我们以字节格式高效执行删除操作。为了简化实施过程，我们使用空的 JSON 对象替换已删除的行，借此快速实现行清除。此项操作只会带来少量存储开销，且消除了对索引内的后续元数据行进行更新的高成本操作需求。要消除空 JSON 对象，我们可以采用离线 vaccum 配合索引更新的方式。

按文件名索引，按索引键分组

在此用例中，我们创建了一个 Amazon DynamoDB 表以存储索引信息。之所以选择 Amazon DynamoDB，是因为其拥有良好的易用性与可扩展性；您可以使用按需计费模型，因此无需猜测可能需要的具体容量单位。在将文件上传至数据湖后，Lambda 函数将解析文件名（例如 1001-.csv）以标记用户标识符，并据此填充 Amazon DynamoDB 元数据表。Userid 为分区键，每个不同存储层都拥有自己的属性。例如，如果用户 1001 在 Amazon S3 及 Amazon RDS 中拥有数据，则其记录将类似于以下形式：

{"userid:": 1001, "s3":{"s3://path1", "s3://path2"}, "RDS":{"db1.table1.column1"}}

复制代码

关于此功能的 Python 实例示例，请参阅 GitHub repo 中的 update-dynamo-metadata.py。

update-dynamo-metadata.py

https://github.com/aws-samples/data-purging-aws-data-lake/blob/master/Scripts/index-by-file-name/update-dynamo-metadata.py

根据删除请求，我们需要查询元数据存储表（即 Amazon DynamoDB）并生成清除报告，该报告中包含关于那些存储层内包含用户记录的详细信息，同时提供有助于加快记录查找速度的其他提示信息。我们将清除报告存储在 Amazon S3 当中。关于实现此逻辑的示例 Lambda 函数，请参阅 GitHub repo 中的 generate-purge-report.py。

generate-purge-report.py

https://github.com/aws-samples/data-purging-aws-data-lake/blob/master/Scripts/index-by-file-name/generate-purge-report.py

在清除获得批准之后，我们将使用清除报告作为输入，借此删除所有对应资源。关于 Lambda 函数的实现示例，请参阅 GitHub repo 中的 gdpr-purge-data.py。

gdpr-purge-data.py

https://github.com/aws-samples/data-purging-aws-data-lake/blob/master/Scripts/index-by-file-name/purge-data.py

实现与技术替代方案

我们探索并评估了多种实现方案，意识到不同的方案各有所长、也都在某些方面有所妥协，包括实现方式的简单性、执行效率、关键数据合规性以及功能完整性等等：

扫描数据文件中的每条记录以创建索引—每次上传文件时，我们都会遍历其记录并生成元组（包含userid, s3Uri,row_number），而后将其插入至我们的元数据存储层内。在删除请求时，我们将获取所请求的用户 ID 的元数据记录，下载相应的 Amazon S3 对象，就地执行删除，而后重新上传经过更新的对象以覆盖现有对象。这是最为灵活的实现方法，因为其支持通过单一对象存储多个用户的数据，也成为目前最为常见的普遍实现方法。但灵活性也有其代价，由于过程中需要下载并重新上传对象，因此删除操作往往会带来网络瓶颈。用户活动数据集（例如客户产品评论）就特别适合使用此种方法，因为各个分区（例如日期分区）中几乎很少出现同一用户发布多条记录的情况，且最好是将多个用户的活动合并到单一文件当中。参考按 Amazon S3 URI 与行号建立索引部分的说明，您可以在 GitHub repo 当中找到相关示例代码。

GitHub repo

https://github.com/aws-samples/data-purging-aws-data-lake/tree/master/Scripts/index-by-row-number

将元数据存储为文件名前缀—在按查询模式定义的不同分区之下，将用户 ID 设定为上传对象的名称前缀，能够帮助我们减少删除请求所需要的搜索操作。元数据处理实用程序能够从文件名中直接查找用户 ID，并相应执行索引维护操作。这种方法能够带来极高的资源清除效率，但每个对象只能对应一个用户，且要求我们将用户 ID 存储在文件名当中，这有可能与信息安全要求相违背。这套方案特别适合管理点击流数据，在此类数据流中，会话期间单一日期分区上的单一用户将产生多个点击事件。根据我们之前在按文件名索引、按索引键分组部分的说明，您可以从 GitHub rep 中下载相关代码库。

GitHub rep

https://github.com/aws-samples/data-purging-aws-data-lake/tree/master/Scripts/index-by-file-name

使用元数据文件—除了上传新对象之外，我们还可以上传可供索引工具使用的元数据文件，借此创建并维护最新索引。根据删除请求，我们可以查询索引、借此将我们指向需要清除的记录位置。此方法最适合在上传新对象时，同步上传对应元数据文件的情况（例如上传多媒体数据）。在其他场景下，在每一次上传对象时都同时上传元数据文件，可能给资源容量带来沉重压力。

使用亚马逊云科技服务的标签功能—每当有新文件被上传至 Amazon S3 时，我们都会使用 Put Object Tagging Amazon S3 操作为用户标识添加键值对。而每当出现用户数据删除请求时，即可使用该标签获取对象并将其删除。使用现有 Amazon S3 API 即可轻松实现这套方案，整个过程相当轻松易行。但这套方案也面临着诸多限制，其假定 Amazon S3 对象与用户之间始终为 1：1 的关系（每个对象仅包含单一用户的数据）；此外，基于标签进行对象搜索的方法效率不高，且将用户标识存储为标签形式的作法也可能有违组织内的信息安全要求。

使用 Apache Hudi—Apache Hudi 已经成为 Amazon S3 之上实现记录层级数据删除功能的一种非常流行的选择。Hudi 的最新版本仅限于 Amazon EMR 使用，因此只适合从零开始构建数据湖的用户，即要求您在创建过程中将数据存储为 Hudi 数据集形式。Hudi 项目本身相当活跃，预计其后续还将迎来更多功能，并与更多亚马逊云科技服务实现集成。

Amazon EMR

http://aws.amazon.com/emr

在具体方法的选择当中，我们始终要求将数据存储层与元数据存储层区分开来。因此，这里提出的各种设计方案具备良好的通用性，能够直接插入任何现有数据管道当中。与选择数据存储层相似，大家在选择存储索引方案时也需要考虑到以下重要因素：

请求并发性—如果不打算同时插入过多请求，您甚至可以考虑直接将 Amazon S3 这类简单存储方案作为初始索引选项。但如果需要面向众多用户处理多项并发写入，则最好选择那些具备更强事务处理能力的服务。

考虑团队的现有专业知识与基础设施—在本文中，我们演示了如何使用 Amazon DynamoDB 与 Amazon RDS Postgres 存储及查询元数据索引。如果您的团队在这方面没有任何经验，而且对 Amazon ES,Amazon DocumentDB(兼容 MongoDB)或者其他存储层的效果基本满意，不妨直接使用。另外，如果您已经拥有一套具备冗余容量的 MySQL 数据库，也可以将其作为索引实现方案以节约运营成本。