Reading and Writing Data

Reading is where a pipeline’s types are decided, and where most of its time goes. This lesson covers the three formats worth using and the one technique that makes file size stop mattering.

CSV, with types declared

suppressPackageStartupMessages({library(tidyverse); library(arrow)})

orders <- read_csv(
  "orders.csv",
  col_types = cols(
    order_id    = col_integer(),
    customer_id = col_integer(),
    ordered_at  = col_date(format = "%Y-%m-%d"),
    status      = col_character(),
    amount      = col_double()
  ),
  na = c("", "NA", "n/a", "NULL")
)

print(orders, n = 3)
cat("problems:", nrow(problems(orders)), "\n")

# A tibble: 6 × 5
  order_id customer_id ordered_at status    amount
     <int>       <int> <date>     <chr>      <dbl>
1     1001           1 2026-01-04 completed  25.5 
2     1002           2 2026-01-05 completed  12   
3     1003           1 2026-01-07 returned   40   
# ℹ 3 more rows
problems: 0

na = is the argument people forget. Without "n/a" in that list, one such value turns the whole amount column into text and every later arithmetic step fails somewhere confusing.

Other options worth knowing:

read_csv("orders.csv", n_max = 1000)                     # sample a big file first
read_csv("orders.csv", skip = 2)                         # junk header lines
read_csv("orders.csv", col_select = c(order_id, amount)) # read fewer columns
read_csv(c("jan.csv", "feb.csv"), id = "source_file")    # many files at once

monthly <- read_csv(c("orders_jan.csv", "orders_feb.csv"),
                    id = "source_file", show_col_types = FALSE)
monthly |> count(source_file) |> print()

# A tibble: 2 × 2
  source_file        n
  <chr>          <int>
1 orders_feb.csv  4120
2 orders_jan.csv  4812

Reading a vector of paths and recording which file each row came from is a two-argument version of a bronze-layer ingest, and source_file answers “where did this row come from” for the life of the table.

Parquet

big <- tibble(
  order_id    = 1:2000000,
  customer_id = sample(1:50000, 2e6, replace = TRUE),
  ordered_at  = as.Date("2026-01-01") + sample(0:89, 2e6, replace = TRUE),
  status      = sample(c("completed","returned","refunded","pending"), 2e6, replace = TRUE),
  amount      = round(runif(2e6, 1, 200), 2),
  note        = paste("order placed via", sample(c("web","app","phone"), 2e6, replace = TRUE))
)

write_csv(big, "big.csv")
write_parquet(big, "big_snappy.parquet", compression = "snappy")
write_parquet(big, "big_zstd.parquet", compression = "zstd")

sizes <- file.info(c("big.csv", "big_snappy.parquet", "big_zstd.parquet"))["size"]
print(round(sizes / 1024^2, 1))

                       size
big.csv               141.4
big_snappy.parquet     44.2
big_zstd.parquet       27.8

bench <- function(label, expr) {
  t <- system.time(result <- force(expr))
  cat(sprintf("%-22s %5.2fs  %s rows\n", label, t[["elapsed"]], format(nrow(result), big.mark = ",")))
}

bench("read_csv",        read_csv("big.csv", show_col_types = FALSE))
bench("read_parquet",    read_parquet("big_snappy.parquet"))
bench("parquet, 2 cols", read_parquet("big_snappy.parquet", col_select = c("status", "amount")))

read_csv                4.81s  2,000,000 rows
read_parquet            0.62s  2,000,000 rows
parquet, 2 cols         0.11s  2,000,000 rows

8× faster than CSV, and 44× when reading two columns of six — a columnar format never touches the columns you did not ask for. Types come back exactly as written, with no guessing and no col_types block.

The rule for pipelines: accept whatever arrives, convert to Parquet as the first transformation, and let everything downstream read that.

Datasets larger than memory

write_dataset(
  big,
  path = "orders_ds",
  format = "parquet",
  partitioning = c("status"),
  compression = "zstd"
)

list.files("orders_ds", recursive = TRUE) |> head(4) |> print()

[1] "status=completed/part-0.parquet" "status=pending/part-0.parquet"  
[3] "status=refunded/part-0.parquet"  "status=returned/part-0.parquet" 

ds <- open_dataset("orders_ds")
print(class(ds))
cat("rows on disk:", format(nrow(ds), big.mark = ","), "\n")

result <- ds |>
  filter(status == "completed", ordered_at >= as.Date("2026-02-01")) |>
  summarise(orders = n(), revenue = sum(amount), .by = ordered_at) |>
  arrange(ordered_at) |>
  collect()

print(result, n = 3)

[1] "FileSystemDataset" "Dataset" "ArrowObject" "R6"
rows on disk: 2,000,000 
# A tibble: 28 × 3
  ordered_at orders  revenue
  <date>      <int>    <dbl>
1 2026-02-01   5602  562884.
2 2026-02-02   5588  561203.
3 2026-02-03   5611  564120.
# ℹ 25 more rows

This is the technique that matters. open_dataset() reads only metadata; the filter and summarise are translated into Arrow operations and pushed down to the files. Nothing enters memory until collect(), and then only the 28-row result.

Compare the two paths:

bench("eager: read then filter",
      read_parquet("big_snappy.parquet") |> filter(status == "completed") |> count())
bench("lazy: dataset pushdown",
      open_dataset("orders_ds") |> filter(status == "completed") |> count() |> collect())

eager: read then filter   0.71s  1 rows
lazy: dataset pushdown    0.04s  1 rows

The lazy version never read the other three partitions. On 2 million rows that is 0.7s versus 0.04s; on 200 million it is the difference between possible and not.

Inspect what will actually run:

open_dataset("orders_ds") |>
  filter(status == "completed", amount > 100) |>
  select(order_id, amount) |>
  show_query()

ExecPlan with 4 nodes:
3:SinkNode{}
  2:ProjectNode{projection=[order_id, amount]}
    1:FilterNode{filter=(amount > 100)}
      0:SourceNode{}

The status filter is absent from the plan because it was satisfied by partition pruning — those directories were never opened.

Not every dplyr verb is supported. When one is not, arrow tells you:

Warning: Expression some_custom_fn(amount) not supported in Arrow; pulling data into R

That warning means the whole dataset is being materialised. On a large one it is the difference between four seconds and running out of memory, so treat it as an error to fix rather than noise.

Writing partitioned output

big |>
  mutate(year_month = format(ordered_at, "%Y-%m")) |>
  write_dataset("orders_by_month", partitioning = "year_month", compression = "zstd")

tibble(file = list.files("orders_by_month", recursive = TRUE, full.names = TRUE)) |>
  mutate(mb = round(file.size(file) / 1024^2, 1)) |>
  print()

# A tibble: 3 × 2
  file                                                  mb
  <chr>                                              <dbl>
1 orders_by_month/year_month=2026-01/part-0.parquet    9.4
2 orders_by_month/year_month=2026-02/part-0.parquet    8.8
3 orders_by_month/year_month=2026-03/part-0.parquet    9.6

Three partitions of about 9 MB. Partition by ordered_at instead and you get 90 files of 300 KB — small enough that per-file overhead outweighs the pruning. Pick a column with few enough distinct values that each partition is worth reading.

Databases

suppressPackageStartupMessages({library(DBI); library(duckdb)})

con <- dbConnect(duckdb(), dbdir = "bookshop.duckdb")
dbWriteTable(con, "orders", big, overwrite = TRUE)

print(dbGetQuery(con, "
  select status, count(*) as orders, round(sum(amount), 2) as revenue
  from orders group by 1 order by revenue desc
"))

dbDisconnect(con, shutdown = TRUE)

     status  orders    revenue
1 completed  500412 50283114.6
2   pending  499802 50194028.2
3  returned  500118 50241880.4
4  refunded  499668 50188442.9

dbWriteTable and dbGetQuery cover most of what a pipeline needs from a database. Lesson 6 covers writing dplyr and having it translated to SQL instead of writing the SQL yourself.

JSON and other formats

library(jsonlite)

nested <- fromJSON('{"order_id": 1001, "customer": {"id": 1, "country": "GB"},
                     "items": [{"sku": "BK-1041", "qty": 1}, {"sku": "BK-2277", "qty": 2}]}')
str(nested, max.level = 2)

flat <- tibble(
  order_id = nested$order_id,
  country  = nested$customer$country,
  items    = list(as_tibble(nested$items))
) |> unnest(items)
print(flat)

List of 3
 $ order_id: int 1001
 $ customer:List of 2
  ..$ id     : int 1
  ..$ country: chr "GB"
 $ items   :'data.frame':	2 obs. of  2 variables:

# A tibble: 2 × 4
  order_id country sku       qty
     <dbl> <chr>   <chr>   <int>
1     1001 GB      BK-1041     1
2     1001 GB      BK-2277     2

fromJSON simplifies arrays of objects into data frames automatically, which is convenient and occasionally surprising — pass simplifyVector = FALSE when you need the raw structure.

For line-delimited JSON, which is what most event feeds produce:

events <- stream_in(file("events.ndjson"), verbose = FALSE) |> as_tibble()
cat("rows:", nrow(events), "\n")

rows: 41209 

Writing safely

write_atomic <- function(data, path) {
  tmp <- paste0(path, ".tmp")
  write_parquet(data, tmp)
  file.rename(tmp, path)
  invisible(path)
}

write_atomic(head(big, 100), "orders_latest.parquet")
cat("exists:", file.exists("orders_latest.parquet"),
    " leftovers:", file.exists("orders_latest.parquet.tmp"), "\n")

exists: TRUE  leftovers: FALSE 

Write to a temporary name and rename, which is atomic on a POSIX filesystem. Without it, a job killed mid-write leaves a truncated Parquet file that every downstream reader fails on — and the failure appears in someone else’s pipeline, not yours.

Practice

read_csv         4.81s
read_parquet     0.62s
parquet, 2 cols  0.11s

ExecPlan with 4 nodes:
    1:FilterNode{filter=(amount > 100)}
      0:SourceNode{}

Warning: Expression my_fn(amount) not supported in Arrow; pulling data into R

# direct write, interrupted
Error in read_parquet("orders.parquet") :
  Invalid: Parquet file size is 0 bytes / Could not open Parquet input source

# atomic write, interrupted
exists: FALSE   leftovers: TRUE

Next: reshaping and joins — pivoting between long and wide, and the joins that lose rows.