数据解决方案的版本控制与测试：在面试风格查询中应用持续集成和单元测试

Image by Author

# 引言

 
每个人都专注于解决问题，但几乎没有人测试解决方案。有时，一个完美运行的脚本可能仅因增加一行新数据或逻辑的微小变化就会崩溃。

在本文中，我们将使用 Python 解决一个特斯拉的面试问题，并通过三个步骤展示版本控制和单元测试如何将一个脆弱的脚本转变为可靠的解决方案。我们将从面试问题开始，最终以使用 GitHub Actions 进行自动化测试而告终。

Image by Author

我们将遵循以下三个步骤使数据解决方案具备生产就绪能力。

首先，我们将解决一个真实的面试问题。其次，我们将添加单元测试，以确保解决方案随着时间的推移保持可靠。最后，我们将使用 GitHub Actions 来自动化测试和版本控制。

# 解决特斯拉的真实面试问题

新产品
计算公司在 2020 年与 2019 年相比推出的产品净变化数量。您的输出应包括公司名称和净差异。
(净差异 = 2020 年推出的产品数量 - 2019 年推出的产品数量。)

在这个 特斯拉的面试问题 中，要求您衡量两年间的产品增长情况。

任务是返回每家公司的名称以及 2020 年和 2019 年产品数量之间的差异。

// 理解数据集

我们首先看一下正在处理的数据集。以下是列名。

让我们预览一下数据集。
 

此数据集包含三列：year、company_name 和 product_name。每一行代表给定年份某公司发布的一款车型。

// 编写 Python 解决方案

我们将使用基本的 pandas 操作来按公司分组、比较并计算产品净变化。我们编写的函数会将数据拆分为 2019 年和 2020 年的子集。

接下来，它按公司名称合并数据，并计算每年独特产品的数量。

import pandas as pd import numpy as np from datetime import datetime df_2020 = car_launches[car_launches['year'].astype(str) == '2020'] df_2019 = car_launches[car_launches['year'].astype(str) == '2019'] df = pd.merge(df_2020, df_2019, how='outer', on=[ 'company_name'], suffixes=['_2020', '_2019']).fillna(0)

最终输出将 2019 年的计数从 2020 年的计数中减去，以得出净差异。这是完整的代码。

import pandas as pd import numpy as np from datetime import datetime df_2020 = car_launches[car_launches['year'].astype(str) == '2020'] df_2019 = car_launches[car_launches['year'].astype(str) == '2019'] df = pd.merge(df_2020, df_2019, how='outer', on=[ 'company_name'], suffixes=['_2020', '_2019']).fillna(0) df = df[df['product_name_2020'] != df['product_name_2019']] df = df.groupby(['company_name']).agg( {'product_name_2020': 'nunique', 'product_name_2019': 'nunique'}).reset_index() df['net_new_products'] = df['product_name_2020'] - df['product_name_2019'] result = df[['company_name', 'net_new_products']]

// 查看预期输出

这是预期输出。
 

# 使用单元测试使解决方案可靠

 
一次解决数据问题并不意味着它会持续有效。新的一行或逻辑调整可能会悄悄地破坏您的脚本。例如，想象一下您不小心在代码中重命名了一个列，将这一行更改为：

df['net_new_products'] = df['product_name_2020'] - df['product_name_2019']

改为这样：

df['new_products'] = df['product_name_2020'] - df['product_name_2019']

逻辑仍然会运行，但您的输出（和测试）会突然失败，因为预期的列名不再匹配。单元测试可以解决这个问题。它们会检查相同输入是否每次都能产生相同输出。如果出现问题，测试会失败并确切指出位置。我们将通过三个步骤来实现这一点，从将面试问题的解决方案转换为函数，到编写一个检查输出是否符合预期的测试。

Image by Author

// 将脚本转换为可重用函数

在编写测试之前，我们需要使解决方案可重用且易于测试。将其转换为函数后，我们可以使用不同的数据集运行它并自动验证输出，而无需每次都重写相同的代码。我们将原始代码更改为一个接受 DataFrame 并返回结果的函数。代码如下。

def calculate_net_new_products(car_launches): df_2020 = car_launches[car_launches['year'].astype(str) == '2020'] df_2019 = car_launches[car_launches['year'].astype(str) == '2019'] df = pd.merge(df_2020, df_2019, how='outer', on=[ 'company_name'], suffixes=['_2020', '_2019']).fillna(0) df = df[df['product_name_2020'] != df['product_name_2019']] df = df.groupby(['company_name']).agg({ 'product_name_2020': 'nunique', 'product_name_2019': 'nunique' }).reset_index() df['net_new_products'] = df['product_name_2020'] - df['product_name_2019'] return df[['company_name', 'net_new_products']]

// 定义测试数据和预期输出

在运行任何测试之前，我们需要知道“正确”的结果是什么。定义预期输出为我们提供了与函数结果进行比较的清晰基准。因此，我们将构建一个小的测试输入并明确定义正确的输出应该是什么。

import pandas as pd # 样本测试数据 test_data = pd.DataFrame({ 'year': [2019, 2019, 2020, 2020], 'company_name': ['Toyota', 'Toyota', 'Toyota', 'Toyota'], 'product_name': ['Camry', 'Avalon', 'Corolla', 'Yaris'] }) # 预期输出 expected_output = pd.DataFrame({ 'company_name': ['Toyota'], 'net_new_products': [0] # 2020 年 2 个 - 2019 年 2 个 })

// 编写和运行单元测试

以下测试代码会检查您的函数是否返回了您期望的完全相同的结果。
 

 

如果不是，测试将失败并告诉您原因，精确到最后一行或一列。

下面的测试使用了上一步中的函数 (calculate_net_new_products()) 和我们定义的预期输出。

import unittest class TestProductDifference(unittest.TestCase): def test_net_new_products(self): result = calculate_net_new_products(test_data) result = result.sort_values('company_name').reset_index(drop=True) expected = expected_output.sort_values('company_name').reset_index(drop=True) pd.testing.assert_frame_equal(result, expected) if __name__ == '__main__': unittest.main()

# 使用持续集成自动化测试

 
编写测试是一个好的开始，但前提是它们确实运行了。您可以在每次代码更改后手动运行测试，但这不具有可扩展性，容易忘记，并且团队成员可能会使用不同的设置。持续集成 (CI) 可以解决这个问题，它在代码推送到存储库时自动运行测试。

GitHub Actions 是一个免费的 CI 工具，它会在每次推送时执行此操作，即使代码、数据或逻辑发生变化，也能保持解决方案的可靠性。它会在每次推送时自动运行您的测试，因此即使代码、数据或逻辑发生变化，您的解决方案也能保持可靠。以下是如何使用 GitHub Actions 应用 CI 的方法。

Image by Author

// 组织项目文件

要将 CI 应用于面试查询，您首先需要将解决方案推送到 GitHub 存储库。（要了解如何创建 GitHub 存储库，请阅读此文）。

然后，设置以下文件：

• solution.py：第 2.1 步中的面试问题解决方案
• expected_output.py：定义第 2.2 步中的测试输入和预期输出
• test_solution.py：使用第 2.3 步中的 unittest 单元测试
• requirements.txt：依赖项（例如 pandas）
• .github/workflows/test.yml：GitHub Actions 工作流文件
• data/car_launches.csv：解决方案使用输入数据集

// 理解存储库布局

这样组织存储库，以便 GitHub Actions 可以在您的 GitHub 存储库中找到它需要的一切，而无需额外的设置。它使事情保持简单、一致，并且对您和他人来说都很容易处理。

my-query-solution/ ├── data/ │ └── car_launches.csv ├── solution.py ├── expected_output.py ├── test_solution.py ├── requirements.txt └── .github/ └── workflows/ └── test.yml

// 创建 GitHub Actions 工作流

现在您已经有了所有文件，最后一个需要的是 test.yml。该文件告诉 GitHub Actions 如何在代码更改时自动运行您的测试。

首先，我们命名工作流并告诉 GitHub 何时运行它。

name: Run Unit Tests on: push: branches: [ main ] pull_request: branches: [ main ]

这意味着每当有人向 main 分支推送代码或打开拉取请求时，测试就会运行。接下来，我们创建一个定义工作流中将发生什么的作业。

jobs: test: runs-on: ubuntu-latest

该作业在 GitHub 的 Ubuntu 环境上运行，每次都会为您提供一个干净的设置。现在我们在该作业中添加步骤。第一个步骤是检出您的存储库，以便 GitHub Actions 可以访问您的代码。

 - name: Checkout repository uses: actions/checkout@v4

然后我们设置 Python 并选择我们想要使用的版本。

 - name: Set up Python uses: actions/setup-python@v5 with: python-version: "3.10"

之后，我们安装 requirements.txt 中列出的所有依赖项。

 - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt

最后，我们运行项目中的所有单元测试。

 - name: Run unit tests run: python -m unittest discover

最后一步会自动运行您的测试，如果出现任何中断，则会显示任何错误。以下是供参考的完整文件：

name: Run Unit Tests on: push: branches: [ main ] pull_request: branches: [ main ] jobs: test: runs-on: ubuntu-latest steps: - name: Checkout repository uses: actions/checkout@v4 - name: Set up Python uses: actions/setup-python@v5 with: python-version: "3.10" - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt - name: Run unit tests run: python -m unittest discover

// 在 GitHub Actions 中查看测试结果

将所有文件上传到 GitHub 存储库后，点击 Actions 选项卡，如下图所示。

点击 Actions 后，如果一切成功运行，您将看到一个绿色的勾号，如下面的屏幕截图所示。
 

 

点击“Update test.yml”查看实际发生的情况。您将获得从设置 Python 到运行测试的完整细分。如果所有测试都通过：

• 每个步骤都会有一个复选标记。
• 这确认了根据您定义的测试，一切都按预期工作。
• 这意味着您的代码在每个阶段都按预期运行，基于您在创建这些测试时设定的目标。
• 输出与您创建测试时设定的目标相匹配。

让我们看看：
 

 

如您所见，我们的单元测试仅用 1 秒就完成了，整个 CI 过程在 17 秒内完成，验证了从设置到测试执行的所有内容。

// 当一个小更改破坏测试时

并非所有的更改都能通过测试。假设您不小心在 solution.py 中重命名了一个列，并将更改发送到 GitHub，例如：

# 原始（运行正常） df['net_new_products'] = df['product_name_2020'] - df['product_name_2019'] # 意外更改 df['new_products'] = df['product_name_2020'] - df['product_name_2019']

让我们看看 Actions 标签中的测试结果。
 

 

我们有一个错误。让我们点击它查看详细信息。
 

 

单元测试没有通过，所以让我们点击“运行单元测试”查看完整的错误消息。
 

 

如您所见，由于函数中的列名不再与测试所期望的名称匹配，我们的测试因 KeyError: 'net_new_products' 而失败。

这就是您如何使代码始终处于检查之下的方式。如果您的团队成员犯了错误，测试将作为您的安全网。

# 使用版本控制来跟踪和测试更改

 
版本控制可以帮助您跟踪所做的每一项更改，无论是在逻辑、测试还是数据集中。假设您想尝试一种新的数据分组方式。不要直接编辑主脚本，而是创建一个新分支：

git checkout -b refactor-grouping

接下来是：

• 进行更改、提交并运行测试。
• 如果所有测试都通过，意味着代码按预期工作，则将其合并。
• 如果未通过，则回滚分支，不影响主代码。

这就是版本控制的强大之处：每一项更改都被跟踪、可测试且可恢复。

# 最终思考

 
大多数人得到正确答案后就停止了。但现实世界中的数据解决方案要求更多。它们奖励那些能够构建经得起时间考验的查询的人，而不仅仅是一次性解决问题的人。

通过版本控制、单元测试和简单的 CI 设置，即使是一次性的面试问题也可以成为您作品集中可靠、可重用的部分。
 
 

Nate Rosidi 是一名数据科学家和产品战略师。他还是教授分析学的兼职教授，并且是 StrataScratch 的创始人，该平台通过来自顶尖公司的真实面试问题帮助数据科学家准备面试。Nate 撰写有关职业市场最新趋势、提供面试建议、分享数据科学项目以及涵盖 SQL 的所有内容。