Ingest CCE-LTER Picoplankton and Bacteria Abundance

Published

2026-08-14

1 Overview

Source: Datazoo table download (https://oceaninformatics.ucsd.edu/datazoo/catalogs/ccelter/datasets/159/datatables/159/download), 16,017 rows, 16 columns (73 cruises, 2004-11-02 to 2023-07-18).

Provider: cce-lter (matches the existing provider group used by ingest_cce-lter_euphausiids/_zoodb/_zooscan – an earlier draft of this notebook used the unhyphenated ccelter, which would have created a duplicate, inconsistent provider group)
Grain: one row per bottle/depth per cast (station-level, multiple depths per cast)
Scope decision: no bounding-box filter applied – source is already scoped to CalCOFI/CCE cruises by the provider. studyName (e.g. 2004-11-02-C-33RR) is the cruise identifier used for key resolution, NOT the Cruise column (which is a separate YYYYMM short code, e.g. 200411 – kept as a raw field but not used for joins).
Completeness: Depth, Heterotrophic Bacteria, Synechococcus, and Picoeukaryotes are >99.9% populated. Prochlorococcus is ~80% populated (3,228/16,017 blank) – plausible non-detection at higher latitudes/depths, not yet confirmed with the provider (see Questions, Q01). Notes is 100% blank in the current export; it is read and normalized to NA rather than dropped, so a future export that starts populating it flows through without a schema change. No literal-zero values appear in any measurement column in this export (confirmed directly against the raw file) – unlike ZooDB, there is no analyzed-but-absent convention to encode; blank simply means not measured.

Code

graph LR
  A[Datazoo table 159<br/>16,017 rows] --> B[rename + type-cast]
  B --> C[picoplankton_bacteria_bottle<br/>position + depth, one row per bottle]
  B --> D[picoplankton_bacteria_measurement<br/>long format: 4 measurement types]
  C -.studyName -> cruise_key.-> E[(shared refs)]
  D -.bottle_id.-> C

graph LR
  A[Datazoo table 159<br/>16,017 rows] --> B[rename + type-cast]
  B --> C[picoplankton_bacteria_bottle<br/>position + depth, one row per bottle]
  B --> D[picoplankton_bacteria_measurement<br/>long format: 4 measurement types]
  C -.studyName -> cruise_key.-> E[(shared refs)]
  D -.bottle_id.-> C

2 Setup

Code

devtools::load_all(here::here("../calcofi4db"))
devtools::load_all(here::here("../calcofi4r"))
librarian::shelf(
  CalCOFI/calcofi4db, CalCOFI/calcofi4r,
  DBI, dplyr, DT, fs, glue, here, htmltools, janitor, jsonlite, knitr,
  lubridate, purrr, readr, sf, stringr, tibble, tidyr, units,
  quiet = T)
options(readr.show_col_types = F)
options(DT.options = list(scrollX = TRUE))

source(here("libs/ingest.R"))
cc <- read_calcofi_meta(here("ingest_cce-lter_picoplankton-bacteria.qmd"))
provider <- cc$provider
dataset <- cc$dataset
dataset_name <- cc$dataset_meta$dataset_name
tables_owned <- cc$tables_owned
dir_label <- glue("{provider}_{dataset}")
dir_parquet <- here(glue("data/parquet/{dir_label}"))
dir_stage <- cc_stage_path("parquet", dir_label, create = TRUE)
db_path <- here(glue("data/wrangling/{dir_label}.duckdb"))
meas_type_csv <- here("metadata/measurement_type.csv")
d_meas_type <- read_measurement_type(meas_type_csv)

if (overwrite) {
  if (file_exists(db_path)) file_delete(db_path)
  if (file_exists(paste0(db_path, ".wal"))) file_delete(paste0(db_path, ".wal"))
  if (dir_exists(paste0(db_path, ".tmp"))) dir_delete(paste0(db_path, ".tmp"))
}
dir_create(dirname(db_path))
con <- get_duckdb_con(db_path)
load_duckdb_extension(con, "spatial")

3 Read Source Data

Fetched reproducibly by libs/download_picoplankton_bacteria.R into {dir_data}/cce-lter/picoplankton-bacteria/. The dataset page links to the Datazoo table download, but that URL redirects to a Login / Accept Data Agreement form and so cannot be fetched unattended; EDI carries the same table (the Datazoo DOI resolves to knb-lter-cce.159), so we pull from there. The revision is deliberately unpinned — this series is ongoing — and the revision actually used is reported below. Columns are renamed per metadata/cce-lter/picoplankton-bacteria/flds_redefine.csv.

Code

source(here("libs/download_picoplankton_bacteria.R"))
pico_csv <- download_picoplankton_bacteria(
  path_expand(glue("{dir_data}/{provider}/{dataset}")),
  overwrite = overwrite_all)

EDI: using cached PicoplanktonandBacteriaAbundance.csv

Code

stopifnot("picoplankton/bacteria CSV not found" = file_exists(pico_csv))

sync_to_gcs(
  local_dir = path_dir(pico_csv),
  gcs_prefix = glue("archive/{provider}/{dataset}"),
  bucket = "calcofi-files-public",
  exclude = c(".DS_Store", "*.tmp", "*.gdoc"))

# A tibble: 0 × 4
# ℹ 4 variables: file <chr>, action <chr>, size <dbl>, reason <chr>

Code

d_raw <- read_csv(pico_csv, col_types = cols(.default = "c"))
cat(glue("Read {format(nrow(d_raw), big.mark=',')} rows, {ncol(d_raw)} columns"), "\n")

Read 16,017 rows, 16 columns

4 Clean, Type-Cast

Canonical names from metadata/cce-lter/picoplankton-bacteria/flds_redefine.csv. studyName (the cruise identifier string, e.g. 2004-11-02-C-33RR) is retained separately from Cruise (the YYYYMM short code) – only the former is used for cruise-key resolution. The four measurement columns (Heterotrophic Bacteria, Prochlorococcus, Synechococcus, Picoeukaryotes) are NOT kept as wide columns – they are pivoted to long format below, following the bottle_measurement/zoodb_measurement pattern used elsewhere in the integrated DB.

Code

d_clean <- d_raw |>
  transmute(
    study_name       = studyName,
    cruise_code      = Cruise,
    datetime_utc     = as_datetime(`Datetime GMT`),
    latitude         = suppressWarnings(as.numeric(`Latitude (º)`)),
    longitude        = suppressWarnings(as.numeric(`Longitude (º)`)),
    line             = suppressWarnings(as.numeric(Line)),
    station          = suppressWarnings(as.numeric(Station)),
    cast_number      = suppressWarnings(as.integer(`Cast Number`)),
    bottle_number    = suppressWarnings(as.integer(`Bottle Number`)),
    assoc_bottle_number = suppressWarnings(as.integer(`Associated Bottle Number`)),
    depth_m          = suppressWarnings(as.numeric(`Depth (m)`)),
    het_bacteria_n_ml    = suppressWarnings(as.numeric(`Heterotrophic Bacteria (number/ml)`)),
    prochlorococcus_n_ml = suppressWarnings(as.numeric(`Prochlorococcus (number/ml)`)),
    synechococcus_n_ml   = suppressWarnings(as.numeric(`Synechococcus (number/ml)`)),
    picoeukaryotes_n_ml  = suppressWarnings(as.numeric(`Picoeukaryotes (number/ml)`)),
    notes            = na_if(Notes, "")) |>
  mutate(
    site_key = if_else(
      is.na(line) | is.na(station), NA_character_,
      sprintf("%05.1f %05.1f", line, station)),
    bottle_id = row_number())

n_all <- nrow(d_clean)
cat(glue("{format(n_all, big.mark=',')} rows read, no bounding-box filter applied ",
         "(source pre-scoped to CalCOFI/CCE cruises)"), "\n")

16,017 rows read, no bounding-box filter applied (source pre-scoped to CalCOFI/CCE cruises)

Code

dbWriteTable(con, "picoplankton_bacteria_bottle", d_clean, overwrite = TRUE)

5 Pivot Measurements to Long Format

Unlike ZooDB, this source has no analyzed-but-absent convention – every blank cell in the raw export is genuinely not measured (confirmed: zero literal 0 values appear in Heterotrophic Bacteria or Prochlorococcus, only blanks). So this pivot simply drops NA, with no explicit-zero retention logic needed.

measurement_type codes are unit-less (het_bacteria, not het_bacteria_n_ml) – units live in measurement_type.csv’s units column, matching the convention used by bottle_measurement/ zoodb_measurement elsewhere in the DB. An earlier draft baked the unit into the type code itself; fixed here.

Code

pb_meas_map <- tribble(
  ~measurement_type,     ~value_col,
  "het_bacteria",         "het_bacteria_n_ml",
  "prochlorococcus",      "prochlorococcus_n_ml",
  "synechococcus",        "synechococcus_n_ml",
  "picoeukaryotes",       "picoeukaryotes_n_ml")

sql_parts <- purrr::pmap_chr(pb_meas_map, function(measurement_type, value_col) {
  glue(
    "SELECT bottle_id, '{measurement_type}' AS measurement_type,
     CAST({value_col} AS DOUBLE) AS measurement_value
     FROM picoplankton_bacteria_bottle WHERE {value_col} IS NOT NULL")
})

sql_create <- glue(
  "CREATE OR REPLACE TABLE picoplankton_bacteria_measurement AS
   SELECT ROW_NUMBER() OVER (ORDER BY bottle_id, measurement_type) AS measurement_id, *
   FROM (
     {paste(sql_parts, collapse = '\nUNION ALL\n')}
   ) sub")
dbExecute(con, sql_create)

[1] 60802

Code

n_meas <- dbGetQuery(con, "SELECT COUNT(*) FROM picoplankton_bacteria_measurement")[[1]]
cat(glue("picoplankton_bacteria_measurement: {format(n_meas, big.mark=',')} rows"), "\n")

picoplankton_bacteria_measurement: 60,802 rows

Code

for (col in pb_meas_map$value_col) {
  tryCatch(
    dbExecute(con, glue('ALTER TABLE picoplankton_bacteria_bottle DROP COLUMN "{col}"')),
    error = function(e) NULL)
}

6 Register Measurement Types

Registers the four new measurement_type codes into the shared metadata/measurement_type.csv and loads the measurement_type table into this connection – an earlier draft referenced measurement_type as a FK target in Schema Documentation / Validate without ever creating it, which would fail. Pattern matches ingest_calcofi_phyllosoma.qmd’s finalize step.

Code

pb_types <- tribble(
  ~measurement_type, ~description,                              ~units,
  "het_bacteria",     "Heterotrophic bacteria abundance (FCM)",  "number/ml",
  "prochlorococcus",  "Prochlorococcus abundance (FCM); ~80% populated, see Q01", "number/ml",
  "synechococcus",    "Synechococcus abundance (FCM)",           "number/ml",
  "picoeukaryotes",   "Picoeukaryote abundance (FCM)",           "number/ml") |>
  mutate(is_canonical = TRUE,
         grain        = "obs",
         `_source_column` = case_when(
           measurement_type == "het_bacteria"    ~ "Heterotrophic Bacteria (number/ml)",
           measurement_type == "prochlorococcus" ~ "Prochlorococcus (number/ml)",
           measurement_type == "synechococcus"   ~ "Synechococcus (number/ml)",
           measurement_type == "picoeukaryotes"  ~ "Picoeukaryotes (number/ml)"),
         `_source_table` = "picoplankton_bacteria_measurement",
         `_source_datasets` = "cce-lter_picoplankton-bacteria",
         `_qual_column` = NA_character_, `_prec_column` = NA_character_)

# upsert so a units/description correction propagates on re-run, and keep the
# registry sorted so its on-disk order is deterministic across ingests
# upsert, not delete-and-replace: the literal below carries no valid_min/
# valid_max, and the naive `filter(!= x) |> bind_rows()` destroyed those
# curated bounds on every re-run (it silently un-declared euphausiid_abundance
# and the picoplankton types mid-release, failing the bounds gate).
d_meas_type <- upsert_measurement_types(d_meas_type, pb_types)
new_types <- pb_types
write_csv(d_meas_type, meas_type_csv, na = "")
dbWriteTable(con, "measurement_type", d_meas_type, overwrite = TRUE)
cat(glue("measurement_type: {nrow(d_meas_type)} types registered ({nrow(new_types)} new)"), "\n")

measurement_type: 200 types registered (4 new)

7 Resolve Ship and Cruise Keys

cruise_key in the shared cruise table is a natural key in format YYYY-MM-NODC, built by the real calcofi4db::derive_cruise_key_on_casts() function (confirmed from ship.R). That function needs a ship_code column (matched against ship.ship_nodc) and a datetime_utc column, and accepts the target table directly via table_name= (confirmed via ingest_calcofi_mets.qmd’s usage) – no need to rename anything to casts. This source has no ship_code natively – only studyName (e.g. 2004-11-02-C-33RR), whose trailing code (33RR) is the ship code embedded in the string – so extract that into a ship_code column, then call the real function against picoplankton_bacteria_bottle directly rather than reimplementing its SQL by hand (an earlier draft did this incorrectly).

Code

load_prior_tables(
  con, parquet_dir = cc_stage_path("parquet", "swfsc_ichthyo"),
  tables = c("ship", "cruise", "grid"), geom_tables = c("grid"), as_view = TRUE)

# A tibble: 3 × 3
  table   rows has_geom
  <chr>  <dbl> <lgl>   
1 cruise   691 FALSE   
2 grid     218 TRUE    
3 ship      48 FALSE

Code

# extract trailing ship code from studyName (e.g. "2004-11-02-C-33RR" -> "33RR")
# NOTE (Q02, resolved): verified against the live release DB for 3 sample
# codes (33RR, 32NM, 31JD); all matched real casts.parquet rows. Still not
# verified across all 73 distinct studyName formats in this dataset.
dbExecute(con, "
  ALTER TABLE picoplankton_bacteria_bottle ADD COLUMN IF NOT EXISTS ship_code VARCHAR")

[1] 0

Code

dbExecute(con, "
  UPDATE picoplankton_bacteria_bottle
  SET ship_code = regexp_extract(study_name, '[0-9A-Z]+$')")

[1] 16017

Code

cruise_key_result <- derive_cruise_key_on_casts(
  con, table_name = "picoplankton_bacteria_bottle", datetime_col = "datetime_utc")

n_ck <- dbGetQuery(con, "SELECT COUNT(*) FROM picoplankton_bacteria_bottle WHERE cruise_key IS NOT NULL")[[1]]
n_all <- dbGetQuery(con, "SELECT COUNT(*) FROM picoplankton_bacteria_bottle")[[1]]
cat(glue("cruise_key match: {n_ck}/{n_all} ({round(100*n_ck/n_all,1)}%)"), "\n")

cruise_key match: 8801/16017 (54.9%)

8 Add Spatial

Code

add_point_geom(con, "picoplankton_bacteria_bottle", lon_col = "longitude", lat_col = "latitude")
assign_grid_key(con, "picoplankton_bacteria_bottle") |> datatable(caption = "Grid assignment")

9 Load Dataset Metadata

Code

d_dataset <- ingest_yaml_to_dataset_df(read_ingest_yaml(here()))
dbWriteTable(con, "dataset", d_dataset, overwrite = TRUE)
cat(glue("dataset: {nrow(d_dataset)} datasets registered"), "\n")

dataset: 16 datasets registered

10 Schema Documentation

Code

pb_rels <- list(
  primary_keys = list(
    picoplankton_bacteria_bottle = "bottle_id",
    picoplankton_bacteria_measurement = "measurement_id",
    measurement_type = "measurement_type"),
  foreign_keys = list(
    list(table = "picoplankton_bacteria_measurement", column = "bottle_id",
         ref_table = "picoplankton_bacteria_bottle", ref_column = "bottle_id"),
    list(table = "picoplankton_bacteria_measurement", column = "measurement_type",
         ref_table = "measurement_type", ref_column = "measurement_type")))

cc_erd(
  con, tables = c("picoplankton_bacteria_bottle", "picoplankton_bacteria_measurement",
                  "measurement_type", "dataset"),
  rels = pb_rels,
  colors = list(lightblue = c("picoplankton_bacteria_bottle", "picoplankton_bacteria_measurement"),
                lightyellow = "measurement_type", white = "dataset"))

Code

build_relationships_json(
  rels = pb_rels, output_dir = dir_parquet, provider = provider, dataset = dataset)

[1] "/Users/bbest/Github/CalCOFI/workflows/data/parquet/cce-lter_picoplankton-bacteria/relationships.json"

11 Validate

Code

results <- validate_for_release(con, checks = "all", strict = FALSE)
cat("Validation:", ifelse(results$passed, "PASSED", "FAILED"), "\n")

Validation: FAILED

Code

if (length(results$errors) > 0)
  cat("Errors:\n", paste("-", results$errors, collapse = "\n"), "\n")

Errors:
 - Table 'picoplankton_bacteria_bottle' has 7216 NULL values in required column 'ship_key'
- Table 'picoplankton_bacteria_bottle' has 7216 NULL values in required column 'cruise_key'

Code

# NULL cruise_key is the expected unmatched remainder pending confirmation of
# study_name -> cruise_key format (see Questions, Q02); not a hard failure.
n_dup <- dbGetQuery(con,
  "SELECT COUNT(*) FROM (SELECT bottle_id, COUNT(*) n FROM picoplankton_bacteria_bottle GROUP BY bottle_id HAVING COUNT(*)>1)")[[1]]
cat(glue("picoplankton_bacteria_bottle bottle_id duplicates: {n_dup}"), "\n")

picoplankton_bacteria_bottle bottle_id duplicates: 0

Code

n_orphan <- dbGetQuery(con,
  "SELECT COUNT(*) FROM picoplankton_bacteria_measurement m
   LEFT JOIN picoplankton_bacteria_bottle b ON m.bottle_id = b.bottle_id
   WHERE b.bottle_id IS NULL")[[1]]
cat(glue("Orphan measurements (no matching bottle): {n_orphan}"), "\n")

Orphan measurements (no matching bottle): 0

Code

n_orphan_type <- dbGetQuery(con,
  "SELECT COUNT(*) FROM picoplankton_bacteria_measurement m
   WHERE m.measurement_type NOT IN (SELECT measurement_type FROM measurement_type)")[[1]]
cat(glue("Orphan measurement_types (not registered): {n_orphan_type}"), "\n")

Orphan measurement_types (not registered): 0

12 Data Preview

Code

cols <- dbGetQuery(con,
  "SELECT column_name FROM information_schema.columns
   WHERE table_name='picoplankton_bacteria_bottle' AND data_type NOT LIKE 'GEOMETRY%'")$column_name
dbGetQuery(con, glue("SELECT {paste(cols, collapse=', ')} FROM picoplankton_bacteria_bottle LIMIT 100")) |>
  datatable(caption = "picoplankton_bacteria_bottle — first 100 rows", rownames = FALSE, filter = "top")

Code

dbGetQuery(con, "SELECT * FROM picoplankton_bacteria_measurement LIMIT 100") |>
  datatable(caption = "picoplankton_bacteria_measurement — first 100 rows", rownames = FALSE, filter = "top")

13 Emit Core Tables

Project this dataset into the shared consolidated core model (design_env-bio-consolidation.md), the same projection release_database.qmd uses to assemble the cross-dataset release. This is a bottle-shaped sample — leaf grain bottle_id, and since the export carries no cast-level event table the bottle is its own root — with an obs headline of the four flow-cytometry counts. Those counts are an environmental measurement vocabulary rather than taxa, so the arm emits realm = 'env' with a NULL taxon_key.

Code

ds_key <- "cce-lter_picoplankton-bacteria"
# no taxa: the four FCM types ARE the measurement vocabulary, not organisms, so
# this dataset has no measurement_taxon rows and emits no taxon/dataset_taxon

# This projection lives here, in the notebook that owns the dataset, not in a
# switch(dataset_key, ...) arm inside calcofi4db. The reusable SHAPES stay in the
# package (sample_arm_self / compat_measurement_sql), so this is a declaration.
append_sample(con, sample_arm_self(
  ds_key, "picoplankton_bacteria_bottle", "bottle_id", "bottle",
  dt_col = "datetime_utc", site_expr = "site_key",
  depth_min = "depth_m", depth_max = "depth_m"))

append_obs(con, glue("
  SELECT 'env', '{ds_key}', {ns_key(ds_key, 'bottle', 'b.bottle_id')},
         b.grid_key, b.cruise_key, b.latitude, b.longitude,
         CAST(b.datetime_utc AS TIMESTAMP), b.depth_m, b.depth_m,
         NULL::VARCHAR, NULL::VARCHAR, m.measurement_type, m.measurement_value,
         NULL::VARCHAR, NULL::DOUBLE
  FROM picoplankton_bacteria_measurement m
  JOIN picoplankton_bacteria_bottle b USING (bottle_id)"))

core <- list(
  sample = dbGetQuery(con, "SELECT COUNT(*) FROM sample")[[1]],
  obs    = dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]])
cat(glue(
  "core projection — sample={core$sample %||% 0} obs={core$obs %||% 0}\n"))

core projection — sample=16017 obs=60802

Code

# every measurement on a bottle must reach obs
n_obs <- dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]]
n_exp <- dbGetQuery(con,
  "SELECT COUNT(*) FROM picoplankton_bacteria_measurement m
   JOIN picoplankton_bacteria_bottle b USING (bottle_id)")[[1]]
stopifnot("obs must be one row per measurement" = n_obs == n_exp)
cat(glue("obs parity: {format(n_obs, big.mark=',')} rows"), "\n")

obs parity: 60,802 rows

Code

stopifnot(
  "every obs.sample_key must resolve in sample" =
    dbGetQuery(con, "SELECT COUNT(*) FROM obs o LEFT JOIN sample s USING (sample_key)
                     WHERE s.sample_key IS NULL")[[1]] == 0)
# serve the retired per-dataset table names as VIEWs over the core, so
# in-notebook consumers and ad-hoc queries keep working against the old names
# (exact for every column the core models, lossy for the rest)
invisible(dbExecute(con, "DROP TABLE IF EXISTS picoplankton_bacteria_measurement_src"))
invisible(dbExecute(con, "ALTER TABLE picoplankton_bacteria_measurement RENAME TO picoplankton_bacteria_measurement_src"))
invisible(dbExecute(con, glue(
  "CREATE OR REPLACE VIEW picoplankton_bacteria_measurement AS
   {compat_measurement_sql(ds_key, 'bottle', 'bottle_id', 'measurement_id')}")))
cat(glue("compat view picoplankton_bacteria_measurement over obs: ",
         "{dbGetQuery(con, 'SELECT COUNT(*) FROM picoplankton_bacteria_measurement')[[1]]} rows"), "\n")

compat view picoplankton_bacteria_measurement over obs: 60802 rows

14 Write Parquet Outputs

Code

dir_create(dir_parquet)
mismatches <- list(cruise_keys = collect_cruise_key_mismatches(con, "picoplankton_bacteria_bottle"))
tbls_out <- core_output_tables(con, extra = c("measurement_type", "dataset"))
parquet_stats <- write_parquet_outputs(
  con = con, output_dir = dir_parquet,
  tables = tbls_out,
  sort_by = list(obs = c("grid_key", "measurement_type")),
  strip_provenance = FALSE, mismatches = mismatches)

build_relationships_json(
  rels = core_relationships(tbls_out), output_dir = dir_parquet,
  provider = provider, dataset = dataset)

[1] "/Users/bbest/Github/CalCOFI/workflows/data/parquet/cce-lter_picoplankton-bacteria/relationships.json"

Code

parquet_stats |> mutate(file = basename(path)) |> select(-path) |>
  datatable(caption = "Parquet export statistics")

15 Write Metadata

Code

d_tbls_rd <- read_csv(here("metadata/cce-lter/picoplankton-bacteria/tbls_redefine.csv"))
d_flds_rd <- read_csv(here("metadata/cce-lter/picoplankton-bacteria/flds_redefine.csv"))
metadata_path <- build_metadata_json(
  con = con, d_tbls_rd = d_tbls_rd, d_flds_rd = d_flds_rd,
  metadata_derived_csv = c(here("metadata/core_dictionary.csv"),
                           here("metadata/cce-lter/picoplankton-bacteria/metadata_derived.csv")),
  output_dir = dir_parquet, tables = tbls_out,
  set_comments = TRUE, provider = provider, dataset = dataset,
  workflow_url = cc$workflow_url, tables_owned = tables_owned)

metadata.json documentation gaps (4 tables, 67 columns) — these render blank in cc_describe_table() / cc_db_catalog():
tables with no description_md: 2    measurement_type, dataset
columns with no description_md: 32    dataset.provider, dataset.dataset, dataset.dataset_name, dataset.dataset_name_short, dataset.category, dataset.color, dataset.description, dataset.citation_main, dataset.citation_others, dataset.link_calcofi_org, dataset.link_data_source, dataset.link_others (+20 more)
measurement columns with no units: 19    dataset.dataset_name_short, dataset.category, dataset.color, dataset.citation_main, dataset.citation_others, dataset.link_calcofi_org, dataset.link_data_source, dataset.link_others, dataset.tables, dataset.coverage_temporal, dataset.coverage_spatial, dataset.license (+7 more)
  backfill via metadata/{provider}/{dataset}/flds_redefine.csv, then re-run

16 Upload to GCS

Code

sync_to_gcs(local_dir = dir_stage, sidecar_dir = dir_parquet, gcs_prefix = glue("ingest/{dir_label}"), bucket = "calcofi-db")

# A tibble: 5 × 4
  file    action    size reason        
  <chr>   <chr>    <dbl> <chr>         
1 <rsync> uploaded    NA parallel rsync
2 <rsync> uploaded    NA parallel rsync
3 <rsync> uploaded    NA parallel rsync
4 <rsync> uploaded    NA parallel rsync
5 <rsync> uploaded    NA parallel rsync

17 Questions for Data Providers

Follow-up questions for CCE-LTER (Michael Landry), ranked by importance. Tracked in metadata/cce-lter/picoplankton-bacteria/questions.csv.

Code

# one validated read + render for every ingest: the vocabulary and the column
# order live in calcofi4db, not in 16 hand-written factor() calls
questions_datatable(
  here(cc$questions_file),
  caption = "Questions for the CCE-LTER picoplankton/bacteria data providers (ranked)")

18 Cleanup

Code

close_duckdb(con)
cat(glue("Parquet outputs written to: {dir_parquet}"), "\n")

Parquet outputs written to: /Users/bbest/Github/CalCOFI/workflows/data/parquet/cce-lter_picoplankton-bacteria