This guide loads company data that arrives as files into Databricks as bronze, silver and gold tables. It covers where the files land, how Auto Loader or COPY INTO ingests them, how typed silver tables stay free of duplicates with MERGE, and how a gold table joins to your own identifiers. The steps suit any licensed feed of files, and the tables and columns are those of the Fokals data dictionary. The examples use Hiring Activity and Job Postings.
Fokals is delivered direct, by REST API and as bulk files, which you load with Auto Loader or COPY INTO. The first task is to pull the data and land it. The delivery page sets out the methods, formats and limits.
Land the files in a Unity Catalog volume
Databricks describes volumes as Unity Catalog objects that govern non-tabular data, and its page on working with files says a volume is read and written through a path of the form /Volumes/catalog/schema/volume/path, with the WRITE VOLUME privilege to write. A landing volume puts the raw files under Unity Catalog grants, as the tables built from them are. Use one folder per table and one sub-folder per pull, and keep each export's manifest beside its files, because the manifest names the period, label versions and licence.
The pull is plain HTTPS: a bearer key per client, rate limits per key by minute and by day, and cursor pagination, with feeds that run oldest first from a time you set. The last cursor is your bookmark for incremental sync, so store it only after the files are written. Keep the key in a secret scope, which Databricks describes as a collection of secrets identified by a name.
import os
import requests
def pull(url: str, dest: str) -> None:
key = dbutils.secrets.get(scope="fokals", key="api-key")
os.makedirs(os.path.dirname(dest), exist_ok=True)
with requests.get(url, headers={"Authorization": f"Bearer {key}"},
stream=True, timeout=600) as r:
r.raise_for_status()
with open(dest, "wb") as f:
for chunk in r.iter_content(chunk_size=1 << 20):
f.write(chunk)
# The export URL comes from the API reference; the folder name carries the pull date.
pull(export_url,
"/Volumes/licensed/landing/fokals/company_hiring_daily/load_date=2026-10-05/part-0001.jsonl")Bronze: ingest the files exactly as they arrive
Databricks's medallion page defines bronze as the raw state of the source in its original formats, and recommends storing most fields as string, VARIANT or binary so that an unexpected schema change drops nothing. Two ingestion tools fit. Databricks's ingestion page points to COPY INTO for files in the order of thousands over time and to Auto Loader for millions or more, to Auto Loader when the schema changes often, and to COPY INTO when you reload a chosen subset of re-uploaded files.
Auto Loader uses the cloudFiles source and keeps what it has discovered in the checkpoint, so that each file is ingested once. Its schema page says it infers every column as a string by default, including nested fields in JSON, and that cloudFiles.inferColumnTypes changes this. The default evolution mode, addNewColumns, makes the stream fail once after it adds new columns. The rescue mode never evolves the schema and puts what does not fit into _rescued_data.
For this feed, choose rescue and plan schema changes deliberately: Fokals announces a breaking change at least 90 days ahead under a new version name. Add schema hints for the count objects. Columns such as by_function, by_country and tech_mentions hold {value: count} objects whose keys vary by company, and without a hint, inference treats each key as a field of a struct.
(spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "json")
.option("cloudFiles.schemaLocation", "/Volumes/licensed/landing/_schemas/company_hiring_daily")
.option("cloudFiles.schemaEvolutionMode", "rescue")
.option("cloudFiles.schemaHints",
"by_function map<string,int>, by_seniority map<string,int>, "
"by_country map<string,int>, by_work_mode map<string,int>, "
"tech_mentions map<string,int>")
.load("/Volumes/licensed/landing/fokals/company_hiring_daily/")
.selectExpr("*",
"_metadata.file_path AS source_file",
"current_timestamp() AS ingested_at")
.writeStream
.option("checkpointLocation", "/Volumes/licensed/landing/_checkpoints/company_hiring_daily")
.trigger(availableNow=True)
.toTable("licensed.bronze.company_hiring_daily")
.awaitTermination())Databricks documents Trigger.AvailableNow as the way to run Auto Loader as a batch job that processes what has arrived and stops, scheduled with Lakeflow Jobs. The hidden _metadata column carries file_path, and it appears only when you select it, which is why the stream selects it explicitly. Keeping the file path and the ingestion time in bronze lets any silver row be traced to the file that brought it.
If the files are CSV, set the reader's header option to true. Its escape option defaults to a backslash. In the Fokals CSV files a list or object is JSON text in a single cell, so open a sample file, see how a quotation mark inside such a cell is written, and set escape to match. JSON Lines avoids the question, because a list or object is a nested value, not text in a cell. Fokals sends sample data on request, with the dictionary and methodology.
COPY INTO is the SQL alternative. Files already loaded are skipped on later runs, and Databricks says this holds even if a file has been modified since, so a corrected file landed under the same name is not reloaded. Land corrections under a new name, or set the force copy option on purpose. The FILES list is limited to 1,000 names.
Databricks says you can create an empty placeholder Delta table and let the first COPY INTO infer its schema with mergeSchema set to true. It adds that such a schemaless table cannot take INSERT INTO or MERGE INTO, so use it for bronze only.
CREATE TABLE IF NOT EXISTS licensed.bronze.company_hiring_daily_csv;
COPY INTO licensed.bronze.company_hiring_daily_csv
FROM '/Volumes/licensed/landing/fokals/company_hiring_daily/'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'escape' = '"')
COPY_OPTIONS ('mergeSchema' = 'true');Silver: types, JSON cells and a key
Databricks's medallion page gives silver the work of schema enforcement, handling nulls and missing values, and deduplication. For Hiring Activity (company_hiring_daily), whose rows are per company and closed UTC day, that means typed columns and a primary key of company_id and day. Databricks recommends liquid clustering for all new tables and says to cluster on the columns most used in filters, so cluster on those two.
For flat count objects, a MAP<STRING, INT> column gives typed access. For objects with mixed values, such as the labels of Job Postings, or arrays of objects such as locations and evidence, the VARIANT type is available from Databricks Runtime 15.4 and recommended over JSON strings by Databricks: convert text with parse_json and read paths with the colon syntax, as in labels:job_function::string.
CREATE TABLE IF NOT EXISTS licensed.silver.company_hiring_daily (
company_id STRING, day DATE, isin STRING, figi STRING,
open_postings INT, new_postings INT, closed_postings INT,
by_function MAP<STRING, INT>, median_salary_usd DOUBLE,
source_file STRING, ingested_at TIMESTAMP
) CLUSTER BY (company_id, day);
MERGE INTO licensed.silver.company_hiring_daily AS t
USING (
SELECT company_id, CAST(day AS DATE) AS day, isin, figi,
CAST(open_postings AS INT) AS open_postings,
CAST(new_postings AS INT) AS new_postings,
CAST(closed_postings AS INT) AS closed_postings,
by_function, CAST(median_salary_usd AS DOUBLE) AS median_salary_usd,
source_file, ingested_at
FROM (
SELECT *, row_number() OVER (PARTITION BY company_id, day
ORDER BY ingested_at DESC) AS rn
FROM licensed.bronze.company_hiring_daily
WHERE ingested_at > (SELECT coalesce(max(ingested_at), TIMESTAMP'1970-01-01 00:00:00')
FROM licensed.silver.company_hiring_daily)
)
WHERE rn = 1
) AS s
ON t.company_id = s.company_id AND t.day = s.day
WHEN NOT MATCHED THEN INSERT *;The MERGE is insert-only, so running it twice adds nothing. The inner query matters. Databricks's merge page says a merge can fail when several source rows match the same target row, and that in the insert-only pattern duplicates inside the new data are inserted. The row_number removes them before the merge.
The high-water mark comes from silver's own ingested_at, so a pipeline that was down for a week catches up without a fixed window. With CSV bronze, by_function arrives as a JSON string, and the source query applies from_json(by_function, 'map<string,int>') in its place.
Two cautions follow. A conflicting duplicate, the same key with different values, is ignored by this merge. Fokals writes daily and weekly rows once and never revises them, so a conflict is a defect to investigate, and a query for keys with more than one distinct set of values in bronze finds it. And a table that describes a changing state is different: Job Postings has one row per posting open at any time in the period, so a posting recurs in later periods with a newer last_seen_at. Keep every period's rows in append-only silver for point-in-time work, and build a current-state table with a MERGE keyed on posting_id that updates only when last_seen_at is newer.
Gold: join to your own identifiers
Every Fokals file carries company_id, the name, and for a listed company or its parent the ticker, exchange, MIC, ISIN, LEI and share-class FIGI. Join to your security master on isin or figi. Private companies carry no ISIN, so an inner join drops them, which is what a securities view wants and a company view does not. The window below sums 28 rows of the hiring dataset, which equals 28 days when a company has a row for every day. Check for gaps before relying on it.
CREATE OR REPLACE VIEW licensed.gold.hiring_momentum AS
SELECT m.security_id, h.day, h.open_postings,
sum(h.new_postings) OVER w AS new_28d,
sum(h.closed_postings) OVER w AS closed_28d
FROM licensed.silver.company_hiring_daily AS h
JOIN acme.reference.security_master AS m ON m.isin = h.isin
WINDOW w AS (PARTITION BY h.company_id ORDER BY h.day
ROWS BETWEEN 27 PRECEDING AND CURRENT ROW);Acme Robotics, the buyer in these examples, and its catalogs are illustrative.
Schedule it and watch it
Lakeflow Jobs can run the pull, bronze, silver and gold steps as tasks in order. Its scheduling page lets you choose a time zone that observes daylight saving time or UTC, and warns that a daylight saving zone can skip or delay a run when the clocks change. Choose UTC: a Fokals day is a closed UTC day, and the table cannot exist before that day has closed.
A file arrival trigger is the alternative for the ingest steps. It watches a volume or an external location and makes a best effort to check every minute. Databricks says that when file events are not enabled on the location, the watched folder can hold up to 10,000 files, so move processed files to an archive volume instead of letting the folder grow.
For monitoring, Databricks recommends Spark's streaming query listener and names the metrics numFilesOutstanding and numBytesOutstanding for the backlog. Add two checks of your own. One is the row count of each load against the period named in the manifest. The other is a count of rows with reconstructed true, which marks a period written more than seven days after it closed.
Keeping it point-in-time
This load gives you tables. They become a point-in-time store when you keep every period's rows and never overwrite them, which the write-once Fokals tables make straightforward. Fokals delivers CSV, JSON and JSON Lines, and the Delta tables are yours to define. The pull, the cursor and the checks around the manifest are the engineering of the job, and what you may do with the data is set by the licence. The guide to governing licensed data with Unity Catalog covers that.
Frequently asked questions
Should I use Auto Loader or COPY INTO for a vendor feed?
Databricks's ingestion page says COPY INTO suits files in the order of thousands over time and Auto Loader suits millions or more, and that Auto Loader is the one to choose when the schema changes often. COPY INTO can be easier to manage when you reload a chosen subset of re-uploaded files. One file per table and period stays far below the millions at which Databricks points to Auto Loader, so a feed that announces schema changes in advance can use either.
How do I load JSON columns from a vendor CSV into Databricks?
Read the cell as a string, then convert it in silver. Use a map type for flat count objects, or a VARIANT for mixed objects and arrays. Check how quotation marks inside a cell are escaped in the file, because the CSV reader's escape option defaults to a backslash. JSON Lines files avoid the problem.
How do I stop a Databricks load from creating duplicates?
Layer three protections. Auto Loader keeps discovered files in its checkpoint, so a file is ingested once. COPY INTO skips files it has already loaded, even modified ones. In silver, an insert-only MERGE on the table's key adds nothing on a rerun, provided you remove duplicates inside the new batch first, because Databricks says duplicates within the new data are inserted.
Where should vendor files land in Databricks?
In a Unity Catalog volume, written through a /Volumes/catalog/schema/volume/path path by a user or service principal with WRITE VOLUME. A volume puts the raw files under Unity Catalog grants like any other governed object. Use one folder per table and one sub-folder per pull, keep the manifest beside the files, and move processed files to an archive volume.
How do I schedule a daily Auto Loader load?
Run Auto Loader with trigger(availableNow=True) as a task in Lakeflow Jobs, which Databricks documents as the way to run it as a batch job. Schedule the job in UTC, because a Fokals day is a closed UTC day, or use a file arrival trigger on the landing volume. Add a check of the row count against the period named in the manifest.
How does Fokals data reach Databricks?
Fokals is delivered direct, by REST API and as bulk files in CSV, JSON or JSON Lines. You call the API or fetch the exports with your key, write the files to a Unity Catalog volume and load them with Auto Loader or COPY INTO, as described here.
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.