背景

如收入自动化系列第二篇博客所述，收入数据管道通过复杂的逻辑转换处理大量数据以确认收入。因此，开发健壮的生产测试和集成策略对此项目阶段的成功至关重要。

现状测试流程在报告生成并发布到数据仓库（Redshift）后使用 Redshift 连接器进行数据同步。这导致数据在数据仓库中可供验证前存在约10小时的延迟。此延迟影响了我们验证变更是否准确反映以及数据是否按要求更新的能力。此外，初始流程涉及手动数据验证，增加了人为错误的风险。

为提高效率并减少手动工作，我们实施了利用"staging 管道"概念的新测试策略。这一改进显著加快了测试过程，因为数据在报告生成后立即可用。这使得管道能在流程早期检测错误。

由于 Yelp 产品实现的独特性，我们在测试管道时面临一些挑战：

• 开发环境数据有限，无法在开发测试期间覆盖生产中出现的各种边缘情况。这在数据管道首次在生产中执行时被发现。
• 我们需要确保在开发区域实施的变更不会影响生产管道中的数据正确性。
• 我们必须找到在发布前使用生产数据测试管道行为的方法。

执行计划

术语表

• 计费收入 - Yelp 向客户开具发票的金额。
• 已实现收入/预估收入 - 当 Yelp 完成购买产品交付时计算的收入。
• 收入期间 - 我们确认产品收入的时间段。
• 收入合同 - 定义服务交付和收入确认条款的合同。
• Redshift 连接器 - 将数据从数据管道流加载到 AWS Redshift 的数据连接器。
• 延迟 - Redshift 连接器处理输入数据所需的时间。
• 数据管道 - 指 Yelp 跨服务流式传输数据的基础设施。

为确保数据正确性，我们遵循以下步骤：

步骤1：Staging 管道设置

我们设置了与生产管道并行的独立数据管道配置，称为 staging 管道。其结果发布到 AWS Glue 表，这使我们能够通过 Redshift Spectrum 立即查询数据，从而解决了 Redshift 连接器延迟问题。当实施新的代码变更时，修改 staging 管道以反映这些变更，并使用生产数据执行。这种方法使我们能够通过轻松比较 staging 和生产结果来在生产环境中验证变更，消除了在生产管道中引入错误时需要回滚变更的需求。

下图说明了并行管道架构：

1
2
3
4
5

[生产管道] ← 并行运行 → [Staging 管道]
     ↓                       ↓
[生产数据存储]           [AWS Glue 表]
                              ↓
                     [Redshift Spectrum 查询]

采用这种方法，生产管道及其数据在新变更验证前保持不变。

下图显示验证后的两个管道：

1

[Staging 管道验证] → [变更批准] → [更新生产管道]

步骤2：测试数据生成

开发环境中的测试数据非常有限，因为 Yelp 有多种产品，在生产中具有不同的边缘情况和收入计算要求，从而增加了两个数据集之间的差距。

我们必须概述生产中可能出现的不同场景和边缘情况，并在开发环境中创建数据来模拟这些场景。一旦通过手动验证和数据完整性检查发现边缘情况（如步骤3所述），就在开发环境中创建新的数据点来复制该场景。

下图说明了此过程：

1

[识别边缘情况] → [创建测试数据] → [开发环境测试] → [验证结果]

由于需要作为管道输入的大量数据库表，此过程非常繁琐，因为它涉及手动创建数据点。因此，对于某些场景的开发和测试，我们严重依赖 staging 管道进行验证。未来，我们计划自动化开发环境中的测试数据创建，如未来改进部分所述。

步骤3：数据完整性检查器

理想情况下，完整性检查器用于比较流程输出与其源数据。在我们的案例中，这是收入合同数据管道结果（包括预估收入）与 Yelp 计费系统产生的计费收入之间的比较。

我们探索了各种指标来提高管道结果的可靠性。一些关键指标包括：

• 匹配发票总收入的合同数量：我们想要验证至少99.99%的合同与计费收入匹配，因为需要这种准确性水平来确保系统可靠性。
• 折扣发票不匹配的合同数量：随着 Yelp 在业务和产品提供层面应用不同的折扣和促销活动，我们必须确保管道预测的折扣应用尽可能接近计费系统。
• 无等效发票的合同行数量：Yelp 的某些产品提供以不同方式计费。有些是手动计费，不一定通过自动化计费流程。我们想要弄清楚管道报告了多少此类合同，以及计费系统生成的发票是否未被自动化管道识别。
• 重复或过去合同行数量：我们还想要确认我们发送了唯一合同，并防止冗余的过去合同进入收入确认系统。

确定所需的报告指标后，开发完整性检查器的后续步骤面临更多挑战：

• 管道使用数据库快照作为其每日来源，即源数据库的每日状态保存在数据湖中，但数据库本身不断变化。
• 产品大多在月末计费。因此，我们必须依赖月度比较来确认数据准确性。
• 收入数据管道设计为每日发送收入确认数据，而理想的真实数据源是 Yelp 计费系统每月产生的收入数据。为解决此问题，我们决定开发月度和每日检查。

月度完整性检查：收入确认管道数据发布到 Redshift 表。通过 SQL 查询比较计费系统和合同管道的数据。确定正确的 SQL 查询是迭代完成的，考虑了基于 Yelp 提供的各种产品的不同边缘情况。

月度检查结果示例格式

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

---- Metrics ----
Number of contracts reviewed: 100000
Gross Revenue: xxxx
Net Revenue: xxxx

Number of contracts matching both gross & net: 99799
Gross Revenue: xxxx
Net Revenue: xxxx

Number of contracts with mismatch discount: 24
Gross Revenue: xx
Net Revenue: xx

Number of contracts with mismatch gross revenue: 6
Gross Revenue: xx

Number of contracts with no equivalent invoice: 1
Gross Revenue: xx
Net Revenue: xx

Number of invoice with no equivalent contract: 10
Gross Revenue: xx
Net Revenue: xx

Line match %: 99.997
Net revenue mismatch difference: 1883.12

每日完整性检查：等到月末确定数据准确性会在实施快速代码修改和数据校正回填时带来挑战。将管道生成的数据发布到 Redshift 数据库在数据可用前会经历超过10小时的延迟。为了在管道运行完成后尽快执行检查，我们分两步实施每日检查：

1. 我们将数据发布到数据湖，并创建 AWS Glue 数据目录表，以便在数据可用后立即从数据存储中读取。
2. 我们在 staging 管道结果上实施 SQL 查询，如 Staging 管道设置部分所述。

如果发现任何差异，我们会迅速确定原因，进行纠正，并重新运行当天的管道。

每日检查结果示例格式

1
2
3
4
5

---- Metrics ----
number of contracts with negative revenue: 0
number of contracts passed or expired: 28
number of contracts with unknown program: 0
number of contracts missing parent category: 66

步骤4：数据格式验证

为支持简化的数据上传流程，我们必须验证上传到第三方系统所需的数据满足系统特定的验证规则。我们面临一些挑战，内部文件数据格式、列映射序列或外部表模式差异导致上传失败。

为避免此类失败，我们开发了模式验证批处理，该批处理使用 REST API 从外部系统检索字段映射，并在进行数据上传前将其与收入数据模式进行比较。

此验证层最小化了失败风险并提高了系统可靠性。以下是检索所需详情的请求/响应示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- 请求
curl -X GET --header "token: RandomBasicToken" "https://yourHost/api/integration/v1/upload/mapping?templatename=SAMPLE_TEMPLATE"

-- 响应
{
    "Dateformat":"MM/DD/YYYY",
    "Mapping":[
        {
            "sample_column_id":1,
            "sample_column_name":"random",
            "sample_column_data_type":"string"
        }
    ]
}

步骤5：数据摄取

我们在集成层开发了上传批处理，该批处理压缩特定报告日期的大型数据文件并将其上传到外部系统。启用了日志记录和监控以及时识别问题和失败，有助于构建更具弹性的流程。

我们实施了两种数据上传方法：

使用 REST API 上传：选择标准的 HTTP 方法是常见做法，易于使用且可扩展。此选项在开发阶段初期探索并用于上传。上传请求/响应示例如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- 请求
curl -X POST https://yourHost/api/integration/v1/upload/file \
-H 'cache-control: no-cache' \
-H 'content-type: multipart/form-data; boundary=----WebKitFormBoundary' \
-H 'templatename: Sample Template' \
-H 'token: <generated token>' \
-F 'file=sample_data_file.csv'

-- 响应
{"Message":"File consumed successfully","Result":{"file_name":"sample_data_file.csv","file_request_id":"1000"},"Status":"Success"}

SFTP 上传：此方法使用 SSH 协议，提供加密数据通道并保护敏感数据免受未经授权的访问。我们为每个环境实施了单独的标志级选项来使用 SFTP 上传功能。

在测试这些上传方法时，我们观察到一些差异，简要讨论如下：

• 可靠性问题：测试显示 REST API 的可用性和性能存在可靠性问题。发现 API 响应不稳定，导致上传不一致和多次重试。
• 文件大小限制：REST API 有每个文件50,000条记录的预定义限制，这导致每天生成约15个文件，月末结算期间约50个文件。通过使用 SFTP 上传方法，我们将文件大小限制增加到500,000到700,000条记录。因此，管道现在仅生成4-5个文件，使上传过程更高效且更易于维护。
• 设置过程：发现 SFTP 上传设置过程与 API 设置相比不太复杂。为了在 Yelp 的多个管道（如收入合同和发票数据管道）中保持标准化和更高效的流程，SFTP 上传方法成为我们的首选方法。

分析 Yelp 的系统和每日数据量需求后，我们决定启用 SFTP 上传作为每日批处理上传过程的主要方法。

步骤6：外部系统支持

我们确认文件已成功每日上传到生产区域的外部收入确认系统。在文件卡在外部 SFTP 服务器上的情况下，我们依赖外部第三方团队在其端解决问题，然后服务才能恢复。这些问题可能由以下几个原因引起：

• SFTP 服务器停机或服务中断
• 文件上传到 SFTP 服务器后外部上传作业触发器失败
• 外部系统中表空间不足

为简化集成流程，必须维护清晰的文档和支持渠道，以实现更快的沟通和问题解决。我们建立了内部指南，用于升级高优先级问题和联系适当的支持团队。

未来改进

随着我们继续增强现有实施，未来的改进机会可能包括：

• 自动化测试数据生成：由于表数量众多和不可预见的边缘情况，手动创建测试数据不切实际。自动化测试数据创建可以减少开发时间并改进整体测试过程。
• 优化数据生成管道：为减少冗余合同，折扣计算可以移动到原始数据生成阶段，然后是资格过滤器。这将增强管道性能并提高报告清晰度。
• 提高可维护性：报告中意外的产品类型可能导致外部系统失败。允许用户包含或排除特定产品类型可以简化新产品上线，帮助调试并防止处理失败。
• 增强完整性检查器：改进折扣匹配可以更好地对齐计费和已实现收入，提高数据准确性。
• 简化第三方数据上传：创建用于数据上传的用户界面将赋予下游团队跨环境的更好控制。包括上传历史跟踪还可以减少手动协调、团队依赖和延迟。

致谢

我们在这个项目上已经走了很长的路，这是一次宝贵且有意义的经历。特别感谢收入确认团队每个成员对此项目的承诺，以及来自商务平台和财务系统团队的所有利益相关者的支持。