Skip to content

Writing Data

Flowfile's writers are Polars-compatible, with added cloud storage integration and visual workflow features.

Polars Compatibility

Local file writers work identically to Polars, plus optional description for visual documentation.

Local File Writing

CSV Files

import flowfile as ff

# Basic usage (same as Polars)
df = ff.read_csv("input.csv")
df.write_csv("output.csv")

# With Flowfile description
df.write_csv("processed_data.csv", description="Save cleaned customer data")

# Polars parameters work identically
df.write_csv(
    "output.csv",
    separator=";",
    encoding="utf-8",
    description="Export with semicolon delimiter"
)

Key Parameters (same as Polars):

  • separator: Field delimiter (default: ,)
  • encoding: File encoding (default: utf-8)

Parquet Files

# Basic usage
df.write_parquet("output.parquet")

# With description and compression
df.write_parquet(
    "compressed_data.parquet",
    description="Save with high compression",
    compression="gzip"
)

Arrow IPC / Feather, NDJSON, and Avro Files

# Arrow IPC / Feather (lazy sink)
df.write_ipc("output.arrow")
df.write_ipc("output.arrow", compression="zstd")

# Newline-delimited JSON (lazy sink)
df.write_ndjson("events.ndjson", compression="gzip")

# Avro (eager write — materialised in the worker)
df.write_avro("output.avro", compression="snappy")

sink_ipc / sink_ndjson aliases are available for the lazy formats.

Compression options (compression=):

  • Parquet: "zstd" (default), "snappy", "gzip", "lz4", "brotli", "uncompressed"
  • IPC/Arrow: "uncompressed" (default), "lz4", "zstd"
  • NDJSON: "uncompressed" (default), "gzip", "zstd"
  • Avro: "uncompressed" (default), "snappy", "deflate"

Core never holds the dataset

Lazy formats (IPC, NDJSON) sink directly from the streaming plan; the eager Avro write and all worker-offloaded writes materialise in the worker process, never in flowfile_core.

Cloud Storage Writing

Flowfile extends writing capabilities with specialized cloud storage writers that integrate with secure connection management.

Unified Cloud Storage Writer

write_to_cloud_storage() is a single entry point for writing any supported format to cloud storage.

import flowfile as ff

# Write Parquet (default format)
ff.write_to_cloud_storage(
    df, "s3://bucket/output.parquet",
    connection_name="my-conn",
)

# Write CSV
ff.write_to_cloud_storage(
    df, "s3://bucket/output.csv",
    file_format="csv",
    connection_name="my-conn",
    delimiter=",",
)

# Write Delta with append mode
ff.write_to_cloud_storage(
    df, "s3://warehouse/my_table",
    file_format="delta",
    connection_name="my-conn",
    write_mode="append",
)

Parameters:

  • df: The LazyFrame to write
  • path: Cloud storage destination path
  • file_format: "csv", "parquet", "json", or "delta" (default: "parquet")
  • connection_name: Name of the stored cloud storage connection
  • delimiter: CSV field separator (default: ;). Only used for CSV
  • encoding: CSV encoding (default: utf8). Only used for CSV
  • compression: Parquet compression: "snappy", "gzip", "brotli", "lz4", "zstd" (default: "snappy"). Only used for Parquet
  • write_mode: "overwrite" (default), "append", "error", "upsert", "update" or "delete". Only used for Delta; see Cloud Storage Write Modes
  • partition_by: Delta partition columns, applied when the write creates the table. Only used for Delta
  • merge_keys: Key columns rows are matched on, required for upsert, update and delete. Only used for Delta
  • track_changes: Turn the table's change feed on. Only used for Delta; see Delta Lake Writing

Recommended Approach

write_to_cloud_storage() is the recommended way to write to cloud storage. The format-specific methods below still work and are useful when you want a more concise call for a known format.

Format-Specific Cloud Writers

Cloud CSV Writing

# Write to S3
df.write_csv_to_cloud_storage(
    "s3://my-bucket/output.csv",
    connection_name="my-aws-connection",
    delimiter=",",
    encoding="utf8",
    description="Export processed data to S3"
)

Parameters:

  • path: Full S3 path including bucket and file name
  • connection_name: Name of configured cloud storage connection
  • delimiter: CSV field separator (default: ;)
  • encoding: File encoding (utf8 or utf8-lossy)

Cloud Parquet Writing

# Write to S3 with compression
df.write_parquet_to_cloud_storage(
    "s3://data-lake/processed/results.parquet",
    connection_name="data-lake-connection",
    compression="snappy",
    description="Save analysis results to data lake"
)

Parameters:

  • path: Full S3 path for the output file
  • connection_name: Name of configured cloud storage connection
  • compression: Compression algorithm (snappy, gzip, brotli, lz4, zstd)

Cloud JSON Writing

# Write JSON to cloud storage
df.write_json_to_cloud_storage(
    "s3://api-data/export.json", 
    connection_name="api-storage",
    description="Export for API consumption"
)

Delta Lake Writing

# Replace the table (the default mode)
df.write_delta(
    "s3://warehouse/customer_dim",
    connection_name="warehouse-connection",
    write_mode="overwrite",
    description="Update customer dimension table"
)

# Append to existing Delta table
new_data.write_delta(
    "s3://warehouse/customer_dim",
    connection_name="warehouse-connection",
    write_mode="append",
    description="Add new customers to dimension"
)

Parameters:

  • path: Cloud storage path of the Delta table — the table's folder, e.g. s3://warehouse/customer_dim
  • connection_name: Name of configured cloud storage connection
  • write_mode: "overwrite" (default), "append", "error", "upsert", "update" or "delete". See Cloud Storage Write Modes
  • partition_by: Partition columns, applied when the write creates the table; later writes must match
  • merge_keys: Key columns rows are matched on, required for upsert, update and delete
  • track_changes: Turn the table's change feed on, so a reader can ask for only what each later write changed. Enable-only: False never turns it off. Not allowed with write_mode="overwrite"
  • description: Optional description for this operation

The tested example upserts into a change-tracked Delta table on S3 twice, then reads back only the rows the second write changed:

# The first write creates the table (version 0) with its change feed on.
day1 = ff.from_dict({"order_id": [1, 2], "status": ["new", "new"]})
day1.write_delta(
    table_path, connection_name="analytics-s3",
    write_mode="upsert", merge_keys=["order_id"], track_changes=True,
)

day2 = ff.from_dict({"order_id": [2, 3], "status": ["shipped", "new"]})
day2.write_delta(
    table_path, connection_name="analytics-s3",
    write_mode="upsert", merge_keys=["order_id"], track_changes=True,
)

changes = ff.scan_delta(table_path, connection_name="analytics-s3", changes_since=0).collect()
print(changes.select("order_id", "status", "_change_type").sort("order_id"))

Catalog Writing

Write data to the Flowfile catalog as managed Delta tables. Available as both a standalone function and a FlowFrame method.

The tested example writes an aggregate to the catalog, then queries it back with SQL:

import flowfile as ff
from flowfile_frame import read_catalog_sql

SALES = "https://raw.githubusercontent.com/edwardvaneechoud/flowfile/main/data/templates/supermarket_sales.csv"

income_by_city = (
    ff.read_csv(SALES)
    .group_by("city")
    .agg(ff.col("gross_income").sum().alias("total_income"))
)

ff.write_catalog_table(
    income_by_city, "docs_sales_by_city", schema=ff.default_schema(), write_mode="overwrite"
)

top_cities = read_catalog_sql(
    "SELECT city, total_income FROM docs_sales_by_city ORDER BY total_income DESC"
)

Standalone Function

import flowfile as ff

# Resolve (or create) the target schema once
schema = ff.CatalogReference("sales", auto_create=True).schema("staging", auto_create=True)

ff.write_catalog_table(
    df, "output_table",
    schema=schema,
    write_mode="upsert",
    merge_keys=["id"],
)

Parameters:

  • df: The LazyFrame to write
  • table_name: Name of the catalog table to write to (required)
  • schema: A SchemaReference identifying the target catalog/schema. Preferred over namespace_id.
  • namespace_full_name: The target schema as a plain "catalog.schema" string, resolved when the flow runs. Exported flow code uses this form
  • write_mode: How to handle existing data (default: "overwrite"). See Catalog Write Modes
  • merge_keys: Column names for merge operations (required for upsert, update, delete)
  • track_changes: Turn change tracking on for the table. Enable-only; rejected with write_mode "overwrite", "virtual" or "scd2"
  • description: Optional description for the table

Catalog handles vs raw IDs

schema= accepts a SchemaReference — a validated, name-based handle that resolves to the underlying namespace id once. Construction fails fast if the catalog or schema doesn't exist (or creates it if auto_create=True). The legacy namespace_id=<int> keyword still works for back-compat, but you can't pass both.

FlowFrame Method

# Convenience: write through the schema handle
schema.write_table(df, "output_table", write_mode="overwrite")

# Or as a method on the FlowFrame
df.write_catalog_table(
    "output_table",
    schema=schema,
    write_mode="overwrite",
)

Returns a new child FlowFrame representing the written data, allowing further chaining.

Database Writing

Write data to a SQL database using a stored connection. Available as a method on FlowFrame.

import flowfile as ff

df = ff.read_csv("data.csv")
df.write_database(
    connection_name="my_db",
    table_name="users",
    schema_name="public",
    if_exists="append",
)

Parameters:

  • connection_name: Name of the stored database connection (required)
  • table_name: Name of the table to write to (required)
  • schema_name: Database schema name (e.g., "public", or "dbo" on SQL Server)
  • if_exists: What to do if the table exists: "append", "replace", or "fail" (default: "append")
  • description: Optional description for this operation

Returns a new child FlowFrame.

DuckDB

Connections with database_type="duckdb" write to a local file. The same if_exists modes apply, and rich types (nested lists/structs, decimals, timestamps) are stored natively rather than text-encoded. A DuckDB file accepts one writer at a time — close other tools using the file while a flow writes to it.

Write Modes

Cloud Storage Write Modes

CSV, Parquet and JSON cloud writes always replace the target file. Delta tables take a write_mode:

Mode Description
overwrite Replace the entire table (default)
error Fail if the table already exists
append Add rows to the existing table
upsert Insert new rows or update existing rows matched by merge_keys
update Update only existing rows matched by merge_keys
delete Delete rows matching merge_keys 
# CSV, Parquet and JSON: no write_mode, the file is replaced
df.write_parquet_to_cloud_storage(
    "s3://bucket/data.parquet",
    connection_name="conn",
)

# Delta: append to the existing table
df.write_delta(
    "s3://warehouse/events",
    connection_name="conn",
    write_mode="append"
)

upsert and update first add any source column the table does not have yet.

Google Cloud Storage

upsert, update, delete and track_changes=True are not supported for gs:// paths. They work on S3 and Azure Data Lake Storage.

Catalog Write Modes

The catalog supports these write modes:

Mode Description
overwrite Replace the entire table
error Fail if the table already exists
append Add rows to the existing table
upsert Insert new rows or update existing rows matched by merge_keys
update Update only existing rows matched by merge_keys
delete Delete rows matching merge_keys
scd2 Track history: end-date changed rows and insert new versions, keyed on merge_keys. See Slowly Changing Dimensions
virtual Create a virtual table — no data written to disk 
# Upsert: insert or update based on merge keys
ff.write_catalog_table(
    df, "customers",
    write_mode="upsert",
    merge_keys=["customer_id"],
)

The tested example writes an SCD2-tracked dimension in two runs and reads back both the active rows and the full history:

import flowfile as ff

customers_day1 = ff.from_dict(
    {"customer_id": [1, 2], "tier": ["free", "pro"], "city": ["Amsterdam", "Berlin"]}
)
ff.write_catalog_table(
    customers_day1, "docs_customers_scd2",
    schema=ff.default_schema(), write_mode="scd2", merge_keys=["customer_id"],
)

customers_day2 = ff.from_dict(
    {"customer_id": [1, 2], "tier": ["pro", "pro"], "city": ["Amsterdam", "Berlin"]}
)
# The write hands the rows back with their surrogate key and validity window attached.
keyed_day2 = ff.write_catalog_table(
    customers_day2, "docs_customers_scd2",
    schema=ff.default_schema(), write_mode="scd2", merge_keys=["customer_id"],
)

current = ff.read_catalog_table("docs_customers_scd2", schema=ff.default_schema(), scd2_view="active")
history = ff.read_catalog_table("docs_customers_scd2", schema=ff.default_schema(), scd2_view="all")

Merge Keys Required

The upsert, update, delete, and scd2 modes require merge_keys to be specified.

scd2 keyword arguments require write_mode='scd2'

write_catalog_table and write_table accept scd2_compare_columns, scd2_full_snapshot, scd2_partition_on_current, scd2_output_mode, and the four scd2_*_column name overrides. Passing any of them with a write_mode other than "scd2" raises an error.

What an scd2 write returns

An scd2 write returns a frame carrying the input's columns followed by the four generated columns, so the surrogate key of the version just written is available downstream. scd2_output_mode picks the rows: "input" (the default) returns the input rows in order, each with its current surrogate key; "changed" returns only the versions this write inserted or end-dated; "current" returns the table's whole current slice. Every other write mode returns the input frame unchanged. See Slowly Changing Dimensions.

Virtual Mode

The virtual write mode creates a catalog entry without materializing data to disk. When the virtual table is read, the producer flow is re-executed on demand. This requires the flow to be registered in the catalog. See Virtual Flow Tables for details.

Connection Requirements

All cloud storage writing requires a configured connection referenced by name. See Cloud Connection Management.

← Previous: Reading Data | Next: Data Types →