I Built a Pipeline to Track the Job Market I'm Trying to Break Into.

我构建了一个管道来追踪我试图进入的就业市场。| Towards Data Science

数据工程

我构建了一个管道来追踪我试图进入的就业市场

使用 dbt 快照构建增量管道，以及那个一路上悄悄误导我的 API。

Ibrahim Salami

2026年10月21日

9分钟阅读

更喜欢在Google上查看我们

分享

文章图片由AI生成

因此，我一直在记录自己完全转向数据工程的旅程，到目前为止，我已经构建了两个管道。第一个是定时运行的GitHub ETL管道，从GitHub仓库提取数据，进行转换并存储到SQLite数据库中。这个项目相对直接，让我学习到了ETL、幂等性以及如何使用cron作业进行调度等概念。

第二个项目更进一步，我构建了一个自动化的RSS管道，从各种RSS源提取技术文章并存储到PostgreSQL数据库中。自动化/编排由Kestra处理。这是我第一次尝试使用编排平台，同时也接触到了Docker，我用它来容器化我的数据管道，以便Kestra能够轻松编排并在AWS云平台上部署。

现在，我正在着手一个新项目。这次，我希望选择一个能让我通过重复之前项目某些方面的过程来巩固记忆，同时学习新技能的项目。我把它称为Level 3。

接下来就是具体实施方式。我认为既然我正在尝试进入数据工程领域，就需要一种方法来跟踪开放职位和行业趋势。因此，我构建了一个管道，其唯一职责就是监控我试图获得的职位：数据工程师、分析工程师、ETL开发人员以及其他相关职位的就业市场。每次运行时，它都会检查是否有人想要雇佣像我正在努力成为的那样的人。

这可能听起来有些奇怪。我还没有作为数据工程师被雇佣，却在这里构建生产级基础设施来追踪一个我尚未拥有的职位的市场情况。但这就是公开学习的要点。即使这个事物部分涉及你自己的求职，你也会构建你希望存在的东西。

本文将涵盖从两个数据源顺利进入本地数据仓库，并实现有效跟踪职位开放和关闭情况的SCD Type 2快照的构建过程。第二部分将介绍如何使用Airflow将所有内容整合在一起。

选择技术栈，并立即遇到障碍

因此，计划是从两个数据源（Hacker News作为简单易用的示例，以及一个具有真实叙事的招聘网站）提取数据，将所有内容存入云数据仓库，使用新工具进行编排，并利用dbt快照来跟踪职位随时间的开放和关闭情况。

在选择编排工具时，我查看了实际招聘广告中出现的内容。Apache Airflow仍然遥遥领先。Kestra、Dagster和Prefect都是真实且正在发展的工具，但Airflow是招聘人员搜索的名称，因此这次我选择学习Airflow，转而使用Airflow，不再使用Kestra。

对于数据仓库，我选择了BigQuery。这是因为BigQuery有免费层级，没有信用卡过期的压力，并且在分析工程的招聘广告中频繁出现。直到我真正尝试设置它之前。

结果发现，如果你身处某些国家，Google Cloud 要求必须预付 30 美元才能关联计费账户。这甚至在 "免费" 等级上也必须完成，才能解锁 BigQuery 的 IAM 和服务账户功能。我不愿意为学习项目承担这笔费用，因此转向了 DuckDB。

说实话，一开始这感觉像是降级。DuckDB 的履历重量无法与 BigQuery 相提并论。但 dbt 抽象了大部分仓库特定的语法，因此我现在学习的数据转换和快照工作，如果将来想对接 BigQuery 或 Snowflake 也能无缝迁移。

Hacker News 与真正理解增量加载

我从 Hacker News 开始，因为它的 API 为每个故事提供了干净且持续递增的 ID，这使它成为增量加载的最简单案例。增量加载本质上是只加载新数据或变更数据，而不是每次都重新加载全部数据。这在需要避免数据管道中重复时非常有用。在项目之前，我构建的每个管道都会每次都拉取全部数据，这在小规模下运行良好，但一旦涉及效率或速率限制就会崩溃，而这基本上是所有实际工作场景的需求。

这里的基本思路是：我想维护一个小型文件来记录已处理的最高故事 ID，只获取高于该 ID 的数据。

python

def load_watermark() -> int:
    if WATERMARK_FILE.exists():
        return json.loads(WATERMARK_FILE.read_text()).get("last_item_id", 0)
    return 0
unseen_ids = [item_id for item_id in new_story_ids if item_id > watermark]

第一次运行拉取了 500 个故事。一分钟后的第二次运行只拉取了 6 个。所以一切正常运作，这非常完美。

不过我需要指出一件事。Hacker News 返回了一个名为 by 的字段来表示故事作者。by 是 SQL 的保留关键字，因此在 DuckDB 中我必须将其用引号标注为 "by" 才能创建表。这其实无需担心，我只是想提一下这个细节。

Remotive 的死胡同

这其实是整篇文章最有价值的教训。

最初我选择 Remotive 作为职位信息源，因为它有文档记录的免费 API，支持搜索和分类参数。我编写了一个提取器，用它查询了六个不同的职位名称（数据工程师、分析工程师、ETL 开发者等）并合并了结果。

然而我发现每个查询都返回完全相同的 15 个职位。无论搜索什么，返回的标题和顺序都完全一致。

我的第一个猜测是速率限制。Remotive 的文档说明每分钟不要超过约 2 次请求，而我连续发起了 6 次请求。我认为这是合理的猜测，因此重建了提取器，改为使用分类过滤器发起单次请求，而不是六次独立搜索。

再次运行后仍然得到相同的 15 个职位。于是我直接对实时 API 发起诊断请求，明确设置 limit=3：

req = Request('https://remotive.com/api/remote-jobs?limit=3', headers={'User-Agent': 'Mozilla/5.0'})
data = json.loads(urlopen(req).read())
print('job-count field:', data.get('job-count'))
print('jobs returned:', len(data.get('jobs', [])))
limit=3

仍然返回了 15 个职位。API 自身的 job-count 字段声称总数就是 15。我发送的所有参数（搜索、分类、limit）都被完全忽略。Remotive 的公共 API 无论你请求什么，都只是返回一个固定的通用样本。

最终，我意识到这并不是我代码中的错误。在确认之前，我通过多种方式从自己的机器上直接与实时API进行比对，确保结果无误。实际上，这意味着我为整个项目选择的第二个数据源已经无法使用，我必须在构建过程中及时调整方向。

转向 RemoteOK

我需要一个真正可用的第二个数据源，因此我尝试了 RemoteOK。这是一个成熟且免费的职位信息API，但在围绕它构建任何东西之前，我想确保它返回的是最新数据，而不是像 Remotive 那样返回的旧样本数据。

RemoteOK 提供了一个标签参数，有助于缩小搜索范围，但它的精确度并不高。一个职位可能有30到40个通用技能标签，因此即使搜索“数据”相关职位，也可能返回诸如“首席法律顾问”之类的职位。这并不是我真正想要的结果。

为了解决这个问题，我将数据提取过程分为两个步骤。首先，我使用 tags=data 参数获取一组可能相关的职位。然后，在 Python 中，我将职位名称与一个更严格的关键词列表进行比对。这是一种简单有效的方法，可以在不完全依赖 RemoteOK 过滤机制的情况下减少噪声。

ROLE_KEYWORDS = [
    "data engineer", "analytics engineer", "etl", "data platform",
    "business intelligence", "big data", "data scientist", "data analyst",
]
def matches_target_role(job: dict) -> bool:
    title = (job.get("position") or "").lower()
    return any(keyword in title for keyword in ROLE_KEYWORDS)

在100个职位中，只有16个真正相关。这很好地提醒我，仅依赖API的过滤功能是不够的。将 RemoteOK 的广义过滤与我在 Python 中的严格关键词检查结合起来，比单独依赖其中任何一种方法都能得到更干净的结果。

监控职位变化：dbt 快照与 SCD Type 2

这才是我真正构建这个项目想要学习的部分。

两个提取器在每次运行时都会将当前状态写入 DuckDB。对于 RemoteOK 来说，加载器每次都会完全替换表，因为该API只告诉你当前开放的职位。仅凭这张表，你永远无法知道某个职位曾经存在过。如果在两次运行之间某个职位信息消失，它就会彻底消失，没有任何记录表明它曾经存在过。

dbt 快照通过将当前表与上次看到的表进行比较，并保留历史记录而不是覆盖它，解决了这个问题。这正是数据工程工作中每个职位描述都会提到但几乎没有入门教程实际演示的“缓慢变化维度”（SCD Type 2）。为了更好地理解 SCD Type 2 的概念，这里有一个可以帮助理解的场景。

想象一个叫 Sarah 的客户，她最初住在芝加哥。几个月后，她搬到了纽约。

如果我们只是在数据库中更新她的城市信息，就会丢失她曾经住在芝加哥的信息。但如果业务需要根据客户购买商品时的居住地分析销售情况，该怎么办？

这时 SCD Type 2 就派上用场了。我们不会覆盖 Sarah 现有的记录，而是保留它，并创建一个包含她更新信息的新记录。我们可以使用日期和当前状态标志来跟踪每个版本的有效时间。

这样，我们就能知道 Sarah 现在住在哪里，而不会丢失她以前居住地的信息。在数据仓库中，这使得分析历史变化成为可能，而不仅仅是查看最新信息。

这是我用来设置快照的代码：

sql

{{ config(
target_schema='snapshots',
unique_key='job_id',
strategy='check',
check_cols=['job_title', 'company', 'tags', 'location', 'salary_min', 'salary_max', 'url'],
hard_deletes='invalidate',
) }}
select * from {{ ref('stg_remoteok_jobs') }}

这里最关键的是设置 hard_deletes='invalidate'。通常 dbt 快照会忽略从源表消失的行。这个设置则告诉它采取相反的处理方式：如果上一次运行中存在的 job ID 在本次运行中缺失，不要静默删除该行，而是用 dbt_valid_to 时间戳标记其结束时间。

在信任这个功能之前，我手动进行了测试。我先用两个假的职位信息运行了一次快照，这两个职位都处于开放状态。随后我模拟其中一个关闭并出现一个新的职位，再次运行快照。

text

('2001', 'Senior Data Engineer',  dbt_valid_from=run1, dbt_valid_to=None)       # 仍然开放
('2002', 'Analytics Engineer II', dbt_valid_from=run1, dbt_valid_to=run2)       # 在两次运行之间关闭
('2003', 'ETL Developer',         dbt_valid_from=run2, dbt_valid_to=None)       # 本次运行新开放

这正是我想要的精确行为，而且 dbt 在没有我编写任何比较逻辑代码的情况下就处理了所有情况。

我还添加了超出 dbt 默认范围的测试。除了标准的非空和唯一性检查外，还有一个自定义测试：如果 RemoteOK 表 ever 返回零行，该测试会发出明显错误，这正是 Remotive 教我需要警惕的静默故障类型。对两个原始数据源还添加了新鲜度检查，如果任一数据源超过24小时未刷新会发出警告。

下一步计划

目前两个数据源都是手动运行，一次执行一个脚本。下一步是将两个提取器和 dbt 命令封装进 Airflow DAG，这意味着在只了解 Kestra 的一年后，需要学习另一个编排工具的思维模型。

一旦这个流程能按计划运行，更有趣的问题就出现了：数据实际展示了什么？哪些职位保持开放时间最长？哪些职位标题变化最快？"数据工程师"和"分析工程师"的职位发布行为是否有差异？这是我最期待撰写的部分，因为这是整个流程从练习转变为映射我自身求职过程的镜子时刻。

这是我在持续记录从系统分析师向数据工程师转型系列文章的一部分。如果你一直关注，感谢你的支持。如果你是第一次阅读本系列文章，前面的文章链接在上方。

Connect with me on LinkedIn , YouTube , and Twitter .