Streaming Large Files

Real bioinformatics datasets can be tens or hundreds of gigabytes. BioLang's streaming API lets you process files of any size in constant memory, using lazy evaluation and parallel execution.

Step 1 — Eager vs Lazy Loading

The read_fastq() function loads an entire file into a table in memory. For small files this is fine, but for a 50 GB FASTQ you would run out of RAM. The fastq() function returns a lazy Stream that yields records one at a time in constant memory.

# Table: loads everything into memory (reusable)
# let reads = read_fastq("data/reads.fastq")  # BAD: 50 GB in RAM

# Stream: constant memory regardless of file size
let stream = fastq("data/reads.fastq")

println(typeof(stream))  # Stream

# Nothing has been read yet. The file is opened, but records are
# only read when you consume the stream.

Step 2 — Consuming a Stream

Each FASTQ stream record is a Record with fields: id, description, seq (DNA), length (Int), and quality (String).

# Count reads without loading them all
let n = fastq("data/reads.fastq")
  |> count()

println(f"Total reads: {n}")  # counted one by one, constant memory

# Take the first N records
let first_100 = fastq("data/reads.fastq")
  |> take(100)
  |> collect()  # materializes into a List

println(f"First 100 reads loaded, {len(first_100)} records")

# `drop` works on a materialized collection. Bound memory with `take` first.
let slice = fastq("data/reads.fastq")
  |> take(1100)
  |> collect()
  |> drop(1000)
  |> take(100)

# Iterate manually with for
let total_bases = 0
for read in fastq("data/reads.fastq") {
  total_bases = total_bases + read.length
}
println(f"Total bases: {total_bases}")

Step 3 — Stream Transformations

Streams support the same map, filter, and flat_map operations as lists, but they execute lazily.

# Filter and transform in a single pass through the file
let gc_values = fastq("data/reads.fastq")
  |> filter(|r| r.length >= 50)
  |> map(|r| gc_content(r.seq))
  |> collect()

println(f"GC values for long reads: {len(gc_values)}")
println(f"Mean GC: {round(mean(gc_values), 4)}")

# Chain multiple filters
let clean = fastq("data/reads.fastq")
  |> filter(|r| r.length >= 50)             # minimum length
  |> filter(|r| gc_content(r.seq) > 0.2)    # GC content check
  |> filter(|r| gc_content(r.seq) < 0.8)    # not too GC-rich
  |> collect()

println(f"Clean reads: {len(clean)}")

# For full quality trimming and filtering, use the file-level builtins:
# trim_quality(input_path, output_path, threshold)
# filter_reads(input_path, output_path, {min_len: 50, min_quality: 25})
# trim_reads(input_path, output_path, {leading: 5, trailing: 5})

Step 4 — Aggregation on Streams

Use reduce with an initial value to compute statistics in a single pass. This is equivalent to a fold — the accumulator is threaded through each record.

# Compute statistics in a single pass using reduce with initial value
let stats = fastq("data/reads.fastq")
  |> reduce(|acc, read| {
    {
      n:           acc.n + 1,
      total_bases: acc.total_bases + read.length,
      total_gc:    acc.total_gc + gc_content(read.seq),
    }
  }, { n: 0, total_bases: 0, total_gc: 0.0 })

println(f"Reads:       {stats.n}")
println(f"Total bases: {stats.total_bases}")
println(f"Mean GC:     {round(stats.total_gc / stats.n, 4)}")

# Collect GC values and summarize
let gc_list = fastq("data/reads.fastq")
  |> map(|r| gc_content(r.seq))
  |> collect()

println(f"Mean GC: {round(mean(gc_list), 4)}")
println(f"GC range: {min(gc_list)} - {max(gc_list)}")

Step 5 — Parallel Processing with par_map

For CPU-intensive operations, par_map distributes a bounded List across multiple threads. Materialize a manageable batch before calling it.

# Process a bounded batch in parallel (default: number of CPU cores)
let results = fastq("data/reads.fastq")
  |> take(10_000)
  |> collect()
  |> par_map(|read| {
    # Each read is processed independently on a worker thread
    let gc = gc_content(read.seq)
    let k = kmers(read.seq, 5)
    { id: read.id, gc: gc, len: read.length, kmer_count: len(k) }
  })
  |> filter(|r| r.gc > 0.3)
  |> collect()

println(f"Processed {len(results)} reads in parallel")

The same pattern works for BAM, but requires a local BAM file:

# requires: huge.bam in working directory
# Parallel BAM processing
# BAM stream records are AlignedRead values with computed properties
let bam_batch = bam("huge.bam")
  |> take(10_000)
  |> collect()

let bam_stats = bam_batch
  |> par_map(|record| {
    {
      mapped:  record.is_mapped,
      qual:    record.mapq,
      paired:  record.is_paired,
    }
  })
  |> reduce(|acc, r| {
    {
      mapped:   acc.mapped + if r.mapped { 1 } else { 0 },
      unmapped: acc.unmapped + if not r.mapped { 1 } else { 0 },
      paired:   acc.paired + if r.paired { 1 } else { 0 },
    }
  }, { mapped: 0, unmapped: 0, paired: 0 })

println(f"Mapped: {bam_stats.mapped}, Unmapped: {bam_stats.unmapped}")

Step 6 — Chunked Processing

Use stream_chunks to break a stream into batches for periodic reporting or batch operations.

# Process in chunks of 10,000 records
let chunks = fastq("data/reads.fastq")
  |> stream_chunks(10_000)
  |> collect()

for chunk in chunks {
  let gc_vals = chunk |> map(|r| gc_content(r.seq))
  println(f"Chunk: {len(chunk)} reads, mean GC {round(mean(gc_vals), 3)}")
}

# Count with periodic progress
let total = 0
let chunks = fastq("data/reads.fastq")
  |> stream_chunks(50_000)
  |> collect()

for chunk in chunks {
  total = total + len(chunk)
  println(f"Processed {total} reads...")
}
println(f"Done. Total: {total} reads")

Step 7 — Logging During a Stream

A mapping function can log an element and then return it unchanged before the next transformation. The separate tee(value, path) builtin writes text to a file; it is not a stream callback.

# Log and forward each read
let results = fastq("data/reads.fastq")
  |> filter(|r| r.length >= 100)
  |> map(|r| {
    println(f"Processing read {r.id}")
    r
  })
  |> map(|r| gc_content(r.seq))
  |> collect()

println(f"Computed GC for {len(results)} reads")

Step 8 — Streaming Other Formats

BioLang's short format readers return streams by default. Use the same map, filter, reduce, collect pipeline on any format.

# requires: alignments.bam in working directory
# Stream BAM files
# BAM records have fields: qname, flag, rname, pos, mapq, cigar, seq, qual
# Plus computed properties: is_mapped, is_paired, is_primary, aligned_length, etc.
let mapped_count = bam("alignments.bam")
  |> filter(|r| r.is_mapped and r.mapq >= 20)
  |> count()

println(f"High-quality mapped reads: {mapped_count}")

The remaining formats use the small tutorial fixtures:

# requires: data/variants.vcf, data/sequences.fasta, data/regions.bed
# Stream VCF files
# VCF records have fields: chrom, pos, id, ref, alt, qual, filter, info
let pass_count = vcf("data/variants.vcf")
  |> filter(|v| v.filter == "PASS")
  |> count()

println(f"PASS variants: {pass_count}")

# Stream FASTA files (for large reference genomes)
let gc_per_chrom = fasta("data/sequences.fasta")
  |> map(|record| {
    { name: record.id, gc: gc_content(record.seq) }
  })
  |> collect()

for entry in gc_per_chrom {
  println(f"{entry.name}: GC = {round(entry.gc, 4)}")
}

# Stream a BED file
# BED records have fields: chrom, start, end, name, score, strand
let total_bases_covered = bed("data/regions.bed")
  |> map(|r| r.end - r.start)
  |> collect()
  |> sum()

println(f"Total bases in BED regions: {total_bases_covered}")

Step 9 — Complete Streaming Pipeline

# streaming_qc.bl — production FASTQ QC with streaming

fn stream_qc(input, output, report_path) {
  println(f"Starting streaming QC on {input}...")

  # Pass 1: compute stats (single pass, constant memory)
  let raw_stats = fastq(input)
    |> reduce(|acc, r| {
      let gc = gc_content(r.seq)
      {
        n:        acc.n + 1,
        bases:    acc.bases + r.length,
        gc_sum:   acc.gc_sum + gc,
        min_len:  min(acc.min_len, r.length),
        max_len:  max(acc.max_len, r.length),
      }
    }, { n: 0, bases: 0, gc_sum: 0.0, min_len: 999999, max_len: 0 })

  println(f"Raw: {raw_stats.n} reads, {raw_stats.bases} bases")

  # Pass 2: trim to an intermediate file, then filter to the requested output
  let trimmed = temp_file()
  trim_quality(input, trimmed, 20)
  filter_reads(trimmed, output, { min_len: 50, min_quality: 25 })
  remove(trimmed)

  # Count clean reads
  let clean_count = fastq(output) |> count()
  let pass_rate = round(clean_count * 100.0 / raw_stats.n, 1)
  println(f"Clean: {clean_count} reads ({pass_rate}%)")

  # Write report
  let report = {
    input:       input,
    output:      output,
    raw_reads:   raw_stats.n,
    raw_bases:   raw_stats.bases,
    clean_reads: clean_count,
    pass_rate:   pass_rate,
    mean_gc:     round(raw_stats.gc_sum / raw_stats.n, 4),
    min_length:  raw_stats.min_len,
    max_length:  raw_stats.max_len,
  }

  write_json(report, report_path)
  println(f"Report saved to {report_path}")
}

stream_qc("data/big.fastq", "results/stream_clean.fastq", "results/stream_qc.json")

# Reading the report back validates that a complete JSON document was written.
let saved_report = read_json("results/stream_qc.json")
println(f"Verified report for {saved_report.raw_reads} raw reads")