Modelling licensed company data with dbt

dbt starts where loading ends: it names the tables that a pipeline has loaded, types their columns, tests the assumptions you made about them and tells you when they stop arriving. This guide sets out a working layout for licensed company data, with the Fokals datasets Hiring Activity, Sales Team Metrics, Sales Pay Benchmarks, Market Series, Employee Headcount, Job Postings, Technology Changes and Company Signals as the example: how to declare them as sources, how to stage them, which tests belong on keys and versions, how to set freshness from the rule each table closes by, and why the write-once tables need no snapshot. The YAML follows the current dbt documentation, with data_tests, config blocks and arguments, and older dbt versions place a few keys differently. Every statement about dbt was checked against the pages linked here on 4 October 2026.

Fokals is delivered direct, by REST API and as bulk files, which you load into your warehouse with its own loader and then point dbt at the tables. The guide to ingesting a vendor API with Fivetran or Airbyte covers one way to load them, and incremental sync covers keeping them current.

What dbt does here, and what it does not

The dbt documentation presents sources as the way to give names and descriptions to tables that your extract and load tools have already filled. Declaring a table as a source lets you select from it with the source() function, test what you assume about it and measure its freshness. dbt does not extract or load data, so for Fokals data the load is yours, from the API or from bulk files in CSV or JSON Lines, and a dbt project starts at the table that load creates.

Three properties of the tables decide how much modelling is left. Daily and weekly tables are written once, after the period closes, and are not revised. Every table that names a company carries company_id and, where the company or its parent is listed, ticker, mic, isin, lei and figi. And a bulk file carries lists and objects as JSON in a single cell. What remains is typing, keys, freshness and the joins to your own tables.

Declare the tables as sources

Put one source, fokals, in a properties file under models/. The source properties page lists the keys: schema and loader at source level, with loader informational, config for loaded_at_field and freshness, and data_tests on tables and columns. The block declares two tables with their keys and their freshness, and later sections add the rest. The schema name is yours.

# models/staging/fokals/_fokals__sources.yml
sources:
  - name: fokals
    description: Licensed Fokals company signals, loaded from the API or from bulk files.
    schema: raw_fokals
    loader: bulk files, CSV
    tables:
      - name: company_hiring_daily
        config:
          loaded_at_field: "cast(day as timestamp)"
          freshness:
            warn_after: {count: 3, period: day}
            error_after: {count: 5, period: day}
        data_tests:
          - dbt_utils.unique_combination_of_columns:
              arguments:
                combination_of_columns: [company_id, day]
        columns:
          - name: company_id
            data_tests: [not_null]
          - name: day
            data_tests: [not_null]
      - name: company_sales_weekly
        config:
          loaded_at_field: "cast(week_start as timestamp)"
          freshness:
            warn_after: {count: 15, period: day}
            error_after: {count: 21, period: day}
        data_tests:
          - dbt_utils.unique_combination_of_columns:
              arguments:
                combination_of_columns: [company_id, week_start]

The freshness documentation says loaded_at_field can be a column or an expression, which is why the daily table can use a cast of day. dbt reads the newest value of that field and compares it with the current time, so the choice of field decides what the check means. The freshness section below explains the choice and the thresholds.

Stage each table once

The dbt staging guidance asks for one staging model per source table that renames, casts and computes simply, with no joins or aggregations, usually materialised as a view and the only place that calls source(). Four jobs belong in the Fokals staging models.

1. Type the time columns. day, week_start and as_of are dates, and observed_at, first_seen_at and closed_at are UTC timestamps.
2. Leave the JSON cells as text and parse them in the model that needs them. by_function, by_country, labels, evidence, before and after are JSON, and the function that parses JSON differs by warehouse.
3. Keep one row per key. A posting that stays open across periods arrives once per period if you load extract by extract, so keep the row with the latest last_seen_at. This is the one place where staging changes the row count, and it does so only to remove copies.
4. Name the baseline. found_on_first_read marks a posting already present at the first observation, so its first_seen_at is the baseline and not its opening date. Stage the negation, so that no downstream model counts a first-read posting as an opening.

-- models/staging/fokals/stg_fokals__job_postings.sql
with source as (
    select * from {{ source('fokals', 'job_postings') }}
),

ranked as (
    select
        *,
        row_number() over (partition by posting_id order by last_seen_at desc) as rn
    from source
),

renamed as (
    select
        posting_id,
        company_id,
        isin,
        title,
        country,
        cast(first_seen_at as timestamp)                 as first_seen_at,
        cast(last_seen_at as timestamp)                  as last_seen_at,
        cast(closed_at as timestamp)                     as closed_at,
        not cast(found_on_first_read as boolean)         as opened_observed,
        label_version
    from ranked
    where rn = 1
)

select * from renamed

Do not key an incremental model on the newest day it already holds. A period written more than seven days after it closed carries reconstructed = true, and it can arrive for a day older than your newest, so the model would never see it. Key the model on your own load time, or re-read a trailing window.

Test the keys, the versions and the joins

The dbt documentation recommends that every model has its primary key tested for duplicates and nulls, and that you test what you assume about source data. For the Fokals tables the keys are the grains that the data dictionary names.

The tests on several columns use dbt_utils.unique_combination_of_columns from the dbt-utils package. Two further kinds of test protect you from change. Labels and scores are produced under named, frozen versions, and a breaking change ships as a new version with at least 90 days' notice, so a test on the version column turns the arrival of a new version into a warning in your own pipeline. A test on an enumerated column such as change in Technology Changes (company_tech_events) does the same for the values the dictionary lists. Add the entries below to the tables list of the same source.

      - name: job_postings
        columns:
          - name: label_version
            data_tests:
              - accepted_values:
                  arguments:
                    values: ['jobs-v1', 'jobs-v2']
                  config:
                    severity: warn
      - name: company_tech_events
        columns:
          - name: change
            data_tests:
              - accepted_values:
                  arguments:
                    values: ['added', 'removed', 'changed']
      - name: listed_securities
        columns:
          - name: figi
            data_tests: [unique, not_null]

The third kind is a join test, and it should warn before it fails. Test that each ISIN in your security master exists in the staged listed_securities, and give the test thresholds so that a handful of misses warns and a collapse fails. The severity documentation describes error_if and warn_if for this.

models:
  - name: security_master
    columns:
      - name: isin
        data_tests:
          - relationships:
              arguments:
                to: ref('stg_fokals__listed_securities')
                field: isin
              config:
                error_if: ">50"
                warn_if: ">0"

The test measures your master against the listed-company index. Whether a security then has company data is a coverage question, and a sample for the companies you track answers it.

Test freshness against the period, not the load

A column that your loader adds, such as _loaded_at, answers whether your pipeline ran. The data's own period answers the question that matters more here, which is whether the latest closed period has arrived. If your loader runs but a day is missing, the first check stays green and the second does not.

A period cannot be fresher than its closing rule allows. A daily row is written after its UTC day closes, so the newest day, read as a timestamp at midnight, is normally one to two days old. A weekly row is written after its Monday to Sunday week closes, so the newest week_start is normally 7 to 14 days old. A market series row is written after its Sunday closes, so the newest as_of is normally 1 to 8 days old.

These are starting points derived from the closing rules. Watch your first deliveries and set each warning just beyond what you see. A daily job runs three commands in order.

1. dbt freshness --resource-type source on dbt v2, or dbt source freshness, which the documentation now calls a legacy command that is still supported.
2. dbt test --select "source:fokals", which runs the tests declared on the sources.
3. dbt build --select staging+, which builds and tests the staging models and everything downstream of them.

Why snapshots are not needed, and the one table that earns one

The dbt documentation describes snapshots as a way to keep the earlier versions of rows in a table that gets overwritten, as type 2 slowly changing dimensions: each version is kept with the dates it was valid. The daily and weekly Fokals tables are never overwritten. Each row is written once after its period closes and is not revised, so the table is already its own history. A snapshot of it would store a second copy of every row, with valid-from and valid-to columns that never close, and add no fact. The event tables, Technology Changes and Company Signals, record changes as dated rows and need no snapshot either. The reasoning behind write-once tables is in why we write each table once.

Mutable tables are another matter. listed_securities carries the status of each listing, including delisted, and a listing's company_id is empty until the listing is linked to a company. The index is refreshed monthly, so a snapshot on the check strategy, run after each refresh, records what you knew about the listed-company index on each date. A back-test that joins on identifiers needs that record.

snapshots:
  - name: snap_fokals__listed_securities
    relation: source('fokals', 'listed_securities')
    config:
      unique_key: figi
      strategy: check
      check_cols: [status, company_id, isin, lei]

job_postings rows also change, as last_seen_at moves and closed_at is set, but each posting carries its own dates, so a snapshot of it is optional.

What the tests tell you

A passing key test says the table has the grain the dictionary names. A passing freshness test says the latest closed period has arrived. Labels are model output produced under named, frozen versions, so the version tests above are the way to track them, and a sample on your own list of companies is the way to measure coverage. The methodology states each label version and how to read each table, and the hiring dataset page lists the tables used above.

Frequently asked questions

The queries and code on this page are examples to adapt. Test them in your own environment before you rely on them.

What this page says about the products it names was checked against their public documentation on 4 October 2026. Product and company names are trademarks of their owners. Fokals is not affiliated with them or endorsed by them.