library(coRpusdb)
library(data.table)
# Choose where DuckDB keeps its extensions (silences its storage message).
options(duckdb.home = file.path(tempdir(), "duckdb_home"))
data_dir <- file.path(tempdir(), "speeches_example")
dir.create(data_dir, showWarnings = FALSE)
db_path <- file.path(data_dir, "speeches.duckdb")This article builds a small database of speeches, step by step, with the functions of coRpusdb. Each step is a decision that a real collection also faces. The records are invented, so that the article runs without any network access.
In a project, db_path is set once in paths_and_packages.R. resolve_db_path() also lets an R option or an environment variable point a run to a copy, to test a change or to read while a collection holds the lock:
db_path <- resolve_db_path(db_path, env_var = "SPEECHES_DUCKDB", option = "speeches.db_path")A schema that documents itself
The schema is a list of CREATE TABLE statements, one per table. The comment after each column is its documentation: write it once, here, and the README and reports are generated from it. Tables have no primary key: DuckDB checks keys eagerly, which breaks the “delete then insert” updates used below, so uniqueness is checked in R before writing.
schema <- list(
document = "
CREATE TABLE IF NOT EXISTS document (
doc_id VARCHAR, -- <source>:<source_id>
source VARCHAR, -- collection the record comes from
source_id VARCHAR, -- identifier in the source
title VARCHAR, -- title as published
date DATE, -- delivery date
speaker_raw VARCHAR, -- speaker(s) as written by the source
collected_on DATE -- date of collection
)",
author = "
CREATE TABLE IF NOT EXISTS author (
doc_id VARCHAR, -- -> document
author_order INTEGER, -- position in the byline
name_raw VARCHAR -- name as written by the source
)"
)
con <- db_connect(db_path)
create_schema(con, schema)
schema_dictionary(schema)
#> table column type description
#> <char> <char> <char> <char>
#> 1: document doc_id VARCHAR <source>:<source_id>
#> 2: document source VARCHAR collection the record comes from
#> 3: document source_id VARCHAR identifier in the source
#> 4: document title VARCHAR title as published
#> 5: document date DATE delivery date
#> 6: document speaker_raw VARCHAR speaker(s) as written by the source
#> 7: document collected_on DATE date of collection
#> 8: author doc_id VARCHAR -> document
#> 9: author author_order INTEGER position in the byline
#> 10: author name_raw VARCHAR name as written by the sourceschema_markdown() writes the data dictionary for the README:
cat(schema_markdown(schema_dictionary(schema)))
document
| Column | Type | Description |
|---|---|---|
doc_id |
VARCHAR | |
source |
VARCHAR | collection the record comes from |
source_id |
VARCHAR | identifier in the source |
title |
VARCHAR | title as published |
date |
DATE | delivery date |
speaker_raw |
VARCHAR | speaker(s) as written by the source |
collected_on |
DATE | date of collection |
author
| Column | Type | Description |
|---|---|---|
doc_id |
VARCHAR | -> document |
author_order |
INTEGER | position in the byline |
name_raw |
VARCHAR | name as written by the source |
Keep what the source returned
A collector first stores what the source returned, as it came, in a raw table. Raw tables are only appended to. When the source adds a field, the column is added rather than the collection stopped. Here, a first batch, then a second one with a new field:
batch_1 <- data.frame(
id = c("sp240101", "sp240115"),
title = c("Monetary policy & the economy", "The euro at 25"),
date = c("2024-01-01", "15/01/2024"),
speakers = c("by Christine Lagarde", "Philip R. Lane and Isabel Schnabel")
)
append_raw_records(con, batch_1, "ecb_speech_raw")
batch_2 <- data.frame(
id = "sp240201",
title = "Inflation: the last mile",
date = "2024-02-01",
speakers = "Christine Lagarde",
language = "en"
)
append_raw_records(con, batch_2, "ecb_speech_raw")
#> ℹ ecb_speech_raw: added column language.
db_query(con, "SELECT id, date, language, collected_at FROM ecb_speech_raw")The dates come in two formats and the title contains an HTML entity. Nothing is corrected in the raw table: correcting is a decision (“categorisation”, in Lemercier and Zalc’s terms), taken in the derived tables and recorded there (Lemercier and Zalc 2019, ch. 3).
Derive the documents, and write them in one transaction
The derived document table harmonises what the raw records say. Identifiers combine the source and its own identifier. They are derived from something stable in the source, never from the order of collection.
raw <- db_query(con, "SELECT * FROM ecb_speech_raw")
document <- raw[, .(
doc_id = make_doc_id("ecb", id),
source = "ecb",
source_id = id,
title = title,
date = fcoalesce(as.IDate(date, "%Y-%m-%d"), as.IDate(date, "%d/%m/%Y")),
speaker_raw = speakers,
collected_on = as.IDate(collected_at)
)]
author <- split_author_string(document$doc_id, document$speaker_raw)
author
#> doc_id author_order name_raw
#> <char> <int> <char>
#> 1: ecb:sp240101 1 Christine Lagarde
#> 2: ecb:sp240115 1 Philip R. Lane
#> 3: ecb:sp240115 2 Isabel Schnabel
#> 4: ecb:sp240201 1 Christine Lagardereplace_documents() writes the documents and their authors in one transaction. For every document of the batch, the rows of each table are first deleted, then the new rows inserted. If anything fails, nothing changes, and a rerun does not create duplicates:
replace_documents(con, list(document = document, author = author),
source = "ecb", script = "collecting_ecb.R")
#> ✔ 3 documents written (document: 3, author: 4).
# The same batch, written again: still three documents and four authors.
replace_documents(con, list(document = document, author = author),
source = "ecb", script = "collecting_ecb.R")
#> ✔ 3 documents written (document: 3, author: 4).
db_query(con, "SELECT (SELECT count(*) FROM document) AS documents,
(SELECT count(*) FROM author) AS authors")Every write is logged, with the number of rows written and replaced. A collector that replaces many more rows than it writes has a problem:
db_query(con, "SELECT script, table_name, n_rows, n_deleted FROM ingestion_log")replace_rows() does the same for one table, selecting the rows to replace by key or by a condition (where = "source = 'ecb'"). It refuses columns that the table does not have, so that a schema change is noticed:
replace_rows(con, "document", data.frame(doc_id = "ecb:x", topic = "inflation"),
keys = "doc_id")
#> Error in `replace_rows()`:
#> ! Column not in table document: topic.Check the database against its schema and its documentation
check_schema() compares the tables with the schema, and check_documented_columns() the schema with the README. Both belong in the project’s sanity_check.R, run after every addition to the database:
check_schema(con, schema)
#> ✔ The database matches the schema.
#> Empty data.table (0 rows and 3 cols): table,column,status
readme <- file.path(data_dir, "README.md")
writeLines(c("| `doc_id` | identifier |", "| `title` | title |"), readme)
check_documented_columns(schema, readme)
#> ! 7 columns not documented in 'README.md'.
#> table column
#> <char> <char>
#> 1: document source
#> 2: document source_id
#> 3: document date
#> 4: document speaker_raw
#> 5: document collected_on
#> 6: author author_order
#> 7: author name_rawDerived outputs carry the version that produced them
A cleaning step writes cleaned titles to a file. Its stamp records the method, the version and a fingerprint of the inputs:
clean_titles <- function(x) normalise_for_matching(x)
cleaned <- document[, .(doc_id, title_key = clean_titles(title))]
cleaned_file <- file.path(data_dir, "title_keys.rds")
saveRDS(cleaned, cleaned_file)
write_stamp(cleaned_file, method = "title_keys", version = "v1",
inputs = content_hash(document$title))
cleaned
#> doc_id title_key
#> <char> <char>
#> 1: ecb:sp240101 monetary policy and economy
#> 2: ecb:sp240115 euro at 25
#> 3: ecb:sp240201 inflation last mileThe next run checks the stamp before reusing the file. If the rule or the input has changed, the stale file is refused rather than silently used:
check_stamp(cleaned_file, method = "title_keys", version = "v2")
#> Error in `check_stamp()`:
#> ! '/tmp/RtmpeRpZPL/speeches_example/title_keys.rds' was not built by the
#> current method.
#> ✖ version v1, expected v2
#> ℹ Rebuild it, or delete it to start afresh.Readers never block the writer
A writing process locks a DuckDB file for every other process. The writer refreshes a snapshot between batches, and readers (reports, analyses, downstream projects) open the database read-only, falling back on the snapshot when it is locked:
db_refresh_snapshot(con, db_path, allow_full_copy = TRUE)
#> ℹ Read-only snapshot refreshed: '/tmp/RtmpeRpZPL/speeches_example/speeches_snapshot.duckdb'
db_disconnect(con)
# A downstream project reads with the path: the connection is closed at once.
db_query(db_path, "SELECT doc_id, date FROM document WHERE date >= ?",
params = list(as.Date("2024-01-15")))
# Filtering on many identifiers: register them as a table, do not paste them.
wanted <- data.frame(doc_id = c("ecb:sp240101", "ecb:sp240201"))
db_query(db_path, "SELECT d.doc_id, d.title FROM document d JOIN wanted USING (doc_id)",
register = list(wanted = wanted))
db_remove_snapshot(db_path)
#> ℹ Read-only snapshot removed.db_connect() also caps DuckDB’s memory (6 GB by default, option coRpusdb.memory_limit): DuckDB’s own default is 80% of the machine’s memory, which is too much next to a language model or another session.
Compare and share copies
db_fingerprint() hashes the content of every table independently of row order. Two copies of a database are compared by exchanging this small table:
fp <- db_fingerprint(db_path)
fp
#> table n_rows hash
#> <char> <num> <char>
#> 1: author 4 40285601826652229330
#> 2: document 3 19914208750728421291
#> 3: ecb_speech_raw 3 25543768612954103161
#> 4: ingestion_log 4 34113756448704160496export_parquet() writes the tables to Parquet files with their schema and row counts, to share with a co-author or archive in an open format. import_parquet() rebuilds a database from them:
con <- db_connect_read_only(db_path)
export_dir <- file.path(data_dir, "export")
export_parquet(con, export_dir, tables = c("document", "author"))
#> ℹ Exporting document
#> ✔ Exporting document [9ms]
#>
#> ℹ Exporting author
#> ✔ Exporting author [9ms]
#>
#> ✔ 2 tables exported to '/tmp/RtmpeRpZPL/speeches_example/export' (0 MB).
db_disconnect(con)
copy <- db_connect(file.path(data_dir, "copy.duckdb"))
import_parquet(copy, export_dir)
#> ✔ 2 tables imported; row counts match the manifest.
compare_fingerprints(db_fingerprint(copy), fp[table %in% c("author", "document")])
#> ✔ All tables identical.
#> table n_rows_here n_rows_other status
#> <char> <num> <num> <char>
#> 1: author 4 4 identical
#> 2: document 3 3 identical
db_disconnect(copy)Check what an export contains before sending it: full texts are often licensed, and metadata can identify people.
Test runs, backups and logs
A change is tested on a few documents before a heavy rerun, and a test never overwrites the real outputs. Projects switch the test mode on with options:
options(speeches.test_mode = TRUE, speeches.max_docs = 2)
output_path(file.path(data_dir, "matches.csv"), prefix = "speeches")
#> [1] "/tmp/RtmpeRpZPL/speeches_example/test/matches.csv"
cap_items(document$doc_id, prefix = "speeches", sample = TRUE)
#> ℹ `speeches.max_docs` is set: 2 of 3 items kept (random draw).
#> [1] "ecb:sp240101" "ecb:sp240115"
options(speeches.test_mode = NULL, speeches.max_docs = NULL)Before rerunning a step that overwrites files, back them up, and compare afterwards:
backup <- backup_files(cleaned_file, reason = "title keys v2", backup_dir = data_dir)
#> ✔ 1 file backed up to '/tmp/RtmpeRpZPL/speeches_example/backup_before_title_keys_v2_2026-11-08'.
saveRDS(cleaned[1:2], cleaned_file)
compare_with_backup(backup)
#> ℹ 0 unchanged, 1 changed, 0 missing.
#> file status
#> <char> <char>
#> 1: /tmp/RtmpeRpZPL/speeches_example/title_keys.rds changedLong steps write their progress to a plain log, readable while the database is locked, and run detached from the session with run_in_background():
log_file <- file.path(data_dir, "logs", "pipeline_log.tsv")
pipeline_log("collect", "start", script = "collecting_ecb.R", log_file = log_file)
pipeline_log("collect", "done", "3 speeches", script = "collecting_ecb.R", log_file = log_file)
pipeline_status(log_file)
#> time script step status message
#> <char> <char> <char> <char> <char>
#> 1: 2026-11-08 17:50:35 collecting_ecb.R collect done 3 speechesReferences
Lemercier, Claire, and Claire Zalc. 2019. Quantitative Methods in the Humanities: An Introduction. Translated by Arthur Goldhammer. University of Virginia Press.