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" or "append" (default: "overwrite"). Only used for Delta

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

# Write Delta table (supports append 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: S3 path for the Delta table
  • connection_name: Name of configured cloud storage connection
  • write_mode: overwrite (replace) or append (add to existing)

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.
  • write_mode: How to handle existing data (default: "overwrite"). See Write Modes
  • merge_keys: Column names for merge operations (required for upsert, update, delete)
  • 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

Overwrite vs Append

# Overwrite existing data (default)
df.write_parquet_to_cloud_storage(
    "s3://bucket/data.parquet",
    connection_name="conn",
    write_mode="overwrite"  # Default for most formats
)

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

Append Mode

For cloud storage, append is only supported for Delta Lake format. Other formats always overwrite.

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"]}
)
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, and the four scd2_*_column name overrides. Passing any of them with a write_mode other than "scd2" raises an error.

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 →