Loading company data into Snowflake from files and an API

This guide loads company data into Snowflake from the two ways Fokals delivers it, bulk files and the REST API, and keeps a history table in step with a daily run. It uses Hiring Activity (company_hiring_daily) from the hiring dataset as the example, and the same landing-then-merge pattern applies to the other tables in the data dictionary, with the key each table needs. Fokals is delivered direct, by REST API and as bulk files, which you load with Snowflake's own COPY INTO from a stage: you pull the data, stage it, load it and schedule the run.

The shape of the pipeline

Land the data first and model it second. A landing table takes the file as delivered, and a MERGE moves the rows you have not seen into the table you query. The split has two reasons. Snowflake's COPY INTO reference does not allow MATCH_BY_COLUMN_NAME together with a SELECT transformation, and loading by header name is what keeps a load stable when columns move. A landing table you can inspect also shows what arrived when a number looks wrong.

The tables below name only the columns the example uses. Add the others from the data dictionary: a column that is in the file and not in the table is ignored, and a column that is in the table and not in the file receives NULL, as Snowflake documents for the option. File paths and names in the SQL are illustrative and follow your own convention.

Stages and file formats

Snowflake loads from a stage, and there are two kinds to choose between.

The PUT reference says the command cannot be run from a worksheet and does not work with external stages, and that it compresses a file with gzip unless you turn that off. The staging guide names the clients that can run it. For an external stage Snowflake recommends a storage integration over embedded credentials. An internal stage suits one job that downloads the files and loads them. An external stage suits an organisation that already keeps vendor files in a bucket it controls, copied there by its own job.

Snowflake's guidance on preparing files aims at 100 to 250 MB compressed, split by line so that no record spans two files. Its load considerations advise staging under logical paths by date. Follow both: give each run its own folder, named for the closed day or the period the files cover.

create database if not exists company_data;
create schema if not exists company_data.fokals;
use schema company_data.fokals;

-- an internal named stage for files you upload with PUT
create stage if not exists landing;

-- CSV with a header row: load by column name, JSON cells enclosed in quotes
create file format if not exists csv_header
  type = csv
  parse_header = true
  field_optionally_enclosed_by = '"'
  error_on_column_count_mismatch = false;   -- a file may carry columns the table lacks

-- one JSON object per line
create file format if not exists jsonl
  type = json;

Bulk CSV files

The data dictionary says files are CSV in UTF-8 with one header row, and that lists and objects are JSON in a single cell. UTF-8 is Snowflake's default encoding, so the first fact needs no option. The header row lets PARSE_HEADER = TRUE and MATCH_BY_COLUMN_NAME load by name, so column order does not matter. The header names are lower case and Snowflake stores unquoted column names in upper case, so match case-insensitively.

A cell that holds JSON contains commas and quotes, which is why the format above encloses fields in double quotes. Test one file before you rely on that.

For a table you have not met yet, look at the file before you design anything. INFER_SCHEMA reads the header and a sample of rows and reports the names and types it would use, every column nullable, and CREATE TABLE ... USING TEMPLATE builds a first landing table from that output. Replace the inferred types where you know better.

select *
from table(infer_schema(
  location    => '@landing/hiring_daily/2026-10-03/',
  file_format => 'csv_header'));

Keep an empty cell as NULL. In the dictionary an empty cell means no value was recorded, and never a zero. Snowflake's EMPTY_FIELD_AS_NULL option is TRUE by default, so a plain load already gives NULL. Do not add a NULL_IF that turns blanks into text or numbers.

Land each JSON cell as text. Snowflake's page on transforming data during a load lists PARSE_JSON among the supported functions, but a transformation cannot be combined with matching by name, so parse in the MERGE with TRY_PARSE_JSON. It returns NULL for text that is not valid JSON, where PARSE_JSON would fail the statement, so count its NULL results against the empty cells and a parse failure cannot pass unseen. Dates, numbers and booleans load into typed columns under the default AUTO formats. The dictionary gives times in UTC, so declare any timestamp column as timestamp_tz.

create table if not exists hiring_daily_landing (
  company_id        varchar,
  company           varchar,
  isin              varchar,
  day               date,
  open_postings     number,
  new_postings      number,
  closed_postings   number,
  by_function       varchar,          -- a JSON object, kept as text for now
  median_salary_usd number(18,2),
  reconstructed     boolean
);

-- from the Snowflake CLI, SnowSQL or a driver, not from a worksheet
put file:///data/fokals/hiring_daily_2026-10-03.csv @landing/hiring_daily/2026-10-03/;

truncate table hiring_daily_landing;

copy into hiring_daily_landing
  from @landing/hiring_daily/2026-10-03/
  file_format = (format_name = 'csv_header')
  match_by_column_name = case_insensitive;

Leave on_error at its bulk default, abort_statement, so a bad file stops the run before the merge. To see every error in a file and not only the first, load once with on_error = continue and read the output of Snowflake's VALIDATE function, select * from table(validate(hiring_daily_landing, job_id => '_last')). Snowflake does not allow VALIDATION_MODE together with MATCH_BY_COLUMN_NAME, so that route is closed here.

JSON Lines from the API

Two sources give you JSON Lines. A bulk export can be requested in that format, and your own loader can write each record of an API page as one line. The feeds are cursor-paginated and run oldest first from a time you set, so the last cursor you received is your bookmark. Snowflake supports newline-delimited JSON and loads each top-level, complete object as a separate row, so a JSON Lines file needs no outer-array option. STRIP_OUTER_ARRAY is for a file that is one JSON array.

create table if not exists signals_raw (src variant);

truncate table signals_raw;

copy into signals_raw
  from @landing/feed/2026-10-04/
  file_format = (format_name = 'jsonl');

-- check what arrived before you rely on a path: expect OBJECT
select typeof(src:by_function) as value_type, count(*) as records
from signals_raw
group by 1;

select src:company_id::varchar        as company_id,
       src:day::date                  as day,
       src:open_postings::number      as open_postings,
       src:by_function.sales::number  as open_sales_postings
from signals_raw;

In a VARIANT, element names are case-sensitive and the column name is not, so write each path exactly as the key is spelt. Snowflake also stores dates and timestamps from JSON as strings, so cast them when you model the table, as the select above does.

Daily increments with MERGE

Daily and weekly rows are written once, after the period closes, and are never changed, and reconstructed marks a period written more than seven days late. For those tables an insert-only MERGE fits: a row is added when its key is new and left alone otherwise, so repeating a run cannot alter history. The key of company_hiring_daily is company_id with day. A weekly table uses week_start in place of day, plus any other grain the dictionary names, such as the topic of an intent score.

The source needs care. Snowflake's MERGE reference says that when the source holds duplicate rows with no match in the target, the target receives one copy of each. Remove duplicates from the source first, as the select distinct below does.

create table if not exists company_hiring_daily (
  company_id        varchar,
  company           varchar,
  isin              varchar,
  day               date,
  open_postings     number,
  new_postings      number,
  closed_postings   number,
  by_function       variant,
  median_salary_usd number(18,2),
  reconstructed     boolean
);

merge into company_hiring_daily as t
using (
  select company_id, company, isin, day, open_postings, new_postings, closed_postings,
         try_parse_json(by_function) as by_function, median_salary_usd, reconstructed
  from (
    select distinct company_id, company, isin, day, open_postings, new_postings,
           closed_postings, by_function, median_salary_usd, reconstructed
    from hiring_daily_landing
  )
) as s
on t.company_id = s.company_id and t.day = s.day
when not matched then insert
  (company_id, company, isin, day, open_postings, new_postings, closed_postings,
   by_function, median_salary_usd, reconstructed)
values
  (s.company_id, s.company, s.isin, s.day, s.open_postings, s.new_postings,
   s.closed_postings, s.by_function, s.median_salary_usd, s.reconstructed);

Snowflake also remembers which files a table has loaded and skips them, but its load metadata expires after 64 days, and truncating a table deletes it, which Snowflake documents as allowing the same files to load again. That is why the landing table is truncated at the start of a run: a failed run can be repeated from the top. Treat file bookkeeping as a convenience and the merge key as the guard. Keep reconstructed in the history table so late periods can be separated when timing matters.

The daily run

The loader runs outside Snowflake, on whatever scheduler you already use, and each run follows the same order.

1. Read the bookmark from a state table, create table if not exists load_state (feed varchar, last_cursor varchar). On the first run use the time you want to start from. A first run is a backfill of the period you request.
2. Pull. For files, download the export covering the period since the last run. For the API, request the feed from the bookmark page by page, write each record as one line, and stay inside the per-minute and per-day limits of your key.
3. PUT the files into the run's stage path.
4. Truncate the landing table and COPY INTO it from that path.
5. MERGE into the history table.
6. Save the new bookmark only after the merge succeeded.

Steps 4 and 5 can run inside Snowflake. A task runs SQL or a stored procedure on a cron schedule and must be resumed after it is created. Snowflake's external network access also lets a function or procedure call an outside endpoint with a stored secret, if you prefer to keep the API call in Snowflake. This guide keeps it in the loader.

Checks and scheduling

Check three things after each run. copy_history shows each file with its status and row_count, though the table function returns up to 14 days. The latest day in the history table should be the day you expected. The period and label versions in the export manifest should match what you loaded.

Snowflake describes Snowpipe as a service for loading micro-batches, and a daily file is a batch that a scheduled COPY INTO loads. Delivery is direct, so your loader decides when a file exists, and the delivery page describes the pull.

For the incremental side, see keeping a warehouse in step with incremental API sync. Once the tables are in, joining them to CRM tables is the next step.

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.