Workflow:Langfuse Langfuse Batch export pipeline
| Knowledge Sources | |
|---|---|
| Domains | LLM_Ops, Data_Export, Blob_Storage |
| Last Updated | 2026-02-14 05:00 GMT |
Overview
End-to-end process for exporting trace, observation, and event data from Langfuse into downloadable files (CSV, JSON, JSONL) stored in S3-compatible blob storage, triggered by user request and processed asynchronously.
Description
This workflow describes how Langfuse handles bulk data export requests. Users initiate exports through the UI, specifying the data table (traces, observations, events, sessions), filters, and output format. The export job is queued for asynchronous processing by the worker service, which streams data from ClickHouse, transforms it into the requested format using Node.js Transform streams, uploads the result to blob storage (S3, Azure Blob, or Google Cloud Storage), generates a time-limited signed download URL, and notifies the user via email when the export is ready.
Usage
Execute this workflow when users need to download large datasets from Langfuse for offline analysis, integration with external tools, or compliance requirements. This supports exporting traces, observations/generations, events, sessions, and dataset items in CSV, JSON, or JSONL format with applied filters.
Execution Steps
Step 1: Export Job Creation
The user triggers an export through the tRPC API, specifying the target table, filters, and output format. The system validates permissions via RBAC, creates a BatchExport record in PostgreSQL with QUEUED status, logs an audit entry, and enqueues the job to the BatchExportQueue with the export ID and project ID.
Key considerations:
- RBAC permissions are checked before allowing export creation
- The export record stores the full query specification for reproducibility
- Audit logging tracks who initiated each export and when
- Users can cancel queued exports before processing begins
Step 2: Job Validation and Setup
The worker picks up the job from the BullMQ queue and validates the export request. It checks that the export record exists, has not been cancelled, and is not older than 30 days (stale job protection). The status is updated to PROCESSING. The query specification is parsed to determine the target table, filters, and format.
Key considerations:
- Exports older than 30 days are rejected with an informative message
- Cancelled exports are skipped gracefully without error
- LangfuseNotFoundError is handled gracefully (job not retried)
- Other errors trigger retry with exponential backoff (8 attempts, 5-second initial delay)
Step 3: Data Streaming from ClickHouse
Based on the target table, the appropriate stream function is invoked to query ClickHouse with the specified filters. Traces, observations, and events each have dedicated stream implementations that handle filter translation, score aggregation, and ClickHouse-specific query optimizations. Comment filters are preprocessed separately due to cross-database requirements.
Key considerations:
- Stream functions use ClickHouse streaming queries for memory-efficient processing
- Score names are fetched and flattened into dynamic columns for trace exports
- Observation-level filters are separated from trace-level filters when appropriate
- Empty result sets (e.g., comment filters matching nothing) produce empty export files
Step 4: Format Transformation
The data stream passes through a format-specific Node.js Transform stream. CSV transformation extracts headers from the first record and escapes special characters. JSON transformation wraps records in an array structure. JSONL transformation outputs one JSON object per line. All transforms yield to the event loop periodically to prevent blocking.
Key considerations:
- CSV headers are dynamically determined from the first record's keys
- CSV escaping handles quotes, delimiters, and newlines within values
- JSONL is the most memory-efficient format for large exports
- Transforms yield control every 50ms to maintain server responsiveness
Step 5: Storage Upload
The transformed data stream is uploaded to blob storage using the StorageService abstraction. The upload uses multipart streaming to handle files of any size without loading them entirely into memory. A signed download URL is generated with a configurable expiration time. The storage backend is determined by environment configuration and supports S3, Azure Blob Storage, and Google Cloud Storage.
Key considerations:
- Multipart upload handles arbitrarily large files with configurable part size
- Server-side encryption (SSE/KMS) is supported for S3 backends
- External endpoint configuration allows generating URLs for different access patterns
- Download link expiration is controlled by BATCH_EXPORT_DOWNLOAD_LINK_EXPIRATION_HOURS
Step 6: Completion Notification
The export record is updated with COMPLETED status, the signed download URL, and the URL expiration timestamp. The system fetches the user who initiated the export and sends a success notification email containing the download link, export name, and user greeting. The email uses a React-rendered HTML template.
Key considerations:
- Email sending requires SMTP configuration (EMAIL_FROM_ADDRESS and SMTP_CONNECTION_URL)
- Missing email configuration is handled gracefully (no email sent)
- Download URLs expire after the configured duration (default: 1 hour)
- Failed exports update the record with ERROR status and error message for debugging