Stream output to a NDJSON file — sink

This writes the output of a query directly to a NDJSON file without collecting it in the R session first. This is useful if the output of the query is still larger than RAM as it would crash the R session if it was collected into R.

Usage

sink_ndjson(
  .data,
  path,
  ...,
  maintain_order = TRUE,
  type_coercion = TRUE,
  predicate_pushdown = TRUE,
  projection_pushdown = TRUE,
  simplify_expression = TRUE,
  slice_pushdown = TRUE,
  no_optimization = FALSE,
  mkdir = FALSE
)

Arguments

.data: A Polars LazyFrame.
path: Output file. Can also be a partition_*() function to export the output to multiple files (see Details section below).
...: Ignored.
maintain_order: Whether maintain the order the data was processed (default is TRUE). Setting this to FALSE will be slightly faster.
type_coercion: Coerce types such that operations succeed and run on minimal required memory (default is TRUE).
predicate_pushdown: Applies filters as early as possible at scan level (default is TRUE).
projection_pushdown: Select only the columns that are needed at the scan level (default is TRUE).
simplify_expression: Various optimizations, such as constant folding and replacing expensive operations with faster alternatives (default is TRUE).
slice_pushdown: Only load the required slice from the scan. Don't materialize sliced outputs level. Don't materialize sliced outputs (default is TRUE).
no_optimization: Sets the following optimizations to FALSE: predicate_pushdown, projection_pushdown, slice_pushdown, simplify_expression. Default is FALSE.
mkdir: Recursively create all the directories in the path.

Value

The input LazyFrame.

Details

Partitioned output

It is possible to export a LazyFrame to multiple files, also called partitioned output. A partition can be determined in several ways:

by key(s): split by the values of keys. The amount of files that can be written is not limited. However, when writing beyond a certain amount of files, the data for the remaining partitions is buffered before writing to the file.
by maximum number of rows: if the number of rows in a file reaches the maximum number of rows, the file is closed and a new file is opened.

These partitioning schemes can be used with the functions partition_by_key() and partition_by_max_size(). See Examples below.

Writing a partitioned output usually requires setting mkdir = TRUE to automatically create the required subfolders.

Examples

# This is an example workflow where sink_ndjson() is not very useful because
# the data would fit in memory. It simply is an example of using it at the
# end of a piped workflow.

# Create files for the NDJSON input and output:
file_ndjson <- tempfile(fileext = ".ndjson")
file_ndjson2 <- tempfile(fileext = ".ndjson")

# Write some data in a CSV file
fake_data <- do.call("rbind", rep(list(mtcars), 1000))
jsonlite::stream_out(fake_data, file(file_ndjson), verbose = FALSE)

# In a new R session, we could read this file as a LazyFrame, do some operations,
# and write it to another NDJSON file without ever collecting it in the R session:
scan_ndjson_polars(file_ndjson) |>
  filter(cyl %in% c(4, 6), mpg > 22) |>
  mutate(
    hp_gear_ratio = hp / gear
  ) |>
  sink_ndjson(path = file_ndjson2)


#----------------------------------------------
# Write a LazyFrame to multiple files depending on various strategies.
my_lf <- as_polars_lf(mtcars)

# Split the LazyFrame by key(s) and write each split to a different file:
out_path <- withr::local_tempdir()
sink_ndjson(my_lf, partition_by_key(out_path, by = c("am", "cyl")), mkdir = TRUE)
fs::dir_tree(out_path)
#> /var/folders/p6/nlmq3k8146990kpkxl73mq340000gn/T//RtmpZRFBYO/file26eb10b2deb1
#> ├── am=0.0
#> │   ├── cyl=4.0
#> │   │   └── 0.jsonl
#> │   ├── cyl=6.0
#> │   │   └── 0.jsonl
#> │   └── cyl=8.0
#> │       └── 0.jsonl
#> └── am=1.0
#>     ├── cyl=4.0
#>     │   └── 0.jsonl
#>     ├── cyl=6.0
#>     │   └── 0.jsonl
#>     └── cyl=8.0
#>         └── 0.jsonl

# Split the LazyFrame by max number of rows per file:
out_path <- withr::local_tempdir()
sink_ndjson(my_lf, partition_by_max_size(out_path, max_size = 5), mkdir = TRUE)
fs::dir_tree(out_path) # mtcars has 32 rows so we have 7 output files
#> /var/folders/p6/nlmq3k8146990kpkxl73mq340000gn/T//RtmpZRFBYO/file26eb4f3a3f07
#> ├── 00000000.jsonl
#> ├── 00000001.jsonl
#> ├── 00000002.jsonl
#> ├── 00000003.jsonl
#> ├── 00000004.jsonl
#> ├── 00000005.jsonl
#> └── 00000006.jsonl