Data Anywhere with Pipelines, Event Notifications, and Workflows

Data on Cloudflare: Ingestion, Events, and Durable Workflows

Data is the foundation of nearly every real application: the database holding user records, the analytics tracking sales and errors, the object storage with Parquet files for data science, or the vector database powering AI features. For Cloudflare, the challenge has been making it easier for developers to get data into the platform, store it, and query it at scale.

To address this, Cloudflare is previewing three new services: Event Notifications for triggering asynchronous work, Pipelines for high-scale streaming ingestion, and Workflows for durable, multi-step execution. Each is designed to fill a gap in the developer platform.

Event Notifications: Reacting to Data Changes

When writing data, you often need to trigger other work in response: processing uploads, updating search indexes, or cleaning up related rows in a database. The first step toward a broader Event Notifications framework is now live for R2 object storage.

You can configure changes to any R2 bucket to write directly to a Queue, which can be consumed reliably by a Worker or by external compute pulling from the queue. This marks the beginning of a wider system planned to include events for:

• Writes to Workers KV namespaces
• Updates to D1 databases, including row changes and triggers
• Deployments to Cloudflare Workers

A single Worker consuming notifications is only one pattern. Consuming events often implies coordinating multiple steps that execute reliably, retry on failure, and avoid unnecessary duplication. Event notifications are the natural trigger for a workflow engine that executes durably.

Pipelines: Streaming Ingestion Without the Plumbing

While R2 supports the S3 API for upload and download, many tools and workloads need more: collecting clickstream data in efficient batches, or partitioning data by customer ID or locale into JSON. The S3 API alone does not cover these ingestion patterns.

Pipelines, an upcoming streaming ingestion service, is designed for this. It aggregates incoming data and writes it directly to R2, with no infrastructure to manage and no concerns about durability or partitioning.

$ wrangler pipelines create clickstream-ingest-prod --batch-size="1MB" --batch-timeout-secs=120 --batch-on-json-key=".merchantId" --destination-bucket="prod-cs-data"

✅ Successfully created new pipeline "clickstream-ingest-prod"
📥 Created endpoints:
➡ HTTPS: https://d458dbe698b8eef41837f941d73bc5b3.pipelines.cloudflarestorage.com/clickstream-ingest-prod
➡ WebSocket: wss://d458dbe698b8eef41837f941d73bc5b3.pipelines.cloudflarestorage.com:8443/clickstream-ingest-prod
➡ Kafka: d458dbe698b8eef41837f941d73bc5b3.pipelines.cloudflarestorage.com:9092 (topic: clickstream-ingest-prod)

Creating a globally scalable ingestion endpoint with Pipelines requires no code. Early designs are protocol-agnostic: HTTP clients can push events, WebSockets can stream them, and existing Kafka producers can be redirected to Pipelines without managing the Kafka cluster.

The longer-term vision includes running transformations over the stream. The platform has a compute layer in Workers that can perform scalable stream processing. An early API sketch of the programming model resembles ideas from Apache Beam or Flink:

export default {    
   // Pipeline handler is invoked when batch criteria are met
   async pipeline(stream: StreamPipeline, env: Env, ctx: ExecutionContext): Promise<StreamingPipeline> {
      // ...
      return stream
         // Type: transform(label: string, transformFunc: TransformFunction): Promise<StreamPipeline>
         // Each transform has a label that is used in metrics to provide
    // per-transform observability and debugging
         .transform("human readable label", (events: Array<StreamEvent>) => {
            return events.map((e) => ...)
         })
         .transform("another transform", (events: Array<StreamEvent>) => {
            return events.map((e) => ...)
         })
         .writeToR2({
            format: "json",
            bucket: "MY_BUCKET_NAME",
            prefix: somePrefix,
            batchSize: "10MB"
         })
   }
}

In this model, a Worker describes a pipeline of transformations (map, reduce, filter) over each subset of events. Steps can call out to other services like D1 or KV to hydrate data or look up values during processing. Scaling is handled automatically based on records-per-second or concurrency settings.

Pipelines enters open beta later in 2024, initially supporting HTTP ingestion with R2 as the destination, though additional sources and sinks are planned.

Workflows: Durable Execution Over Data

As data and AI platforms grow, developers need reliable, repeatable workflows that operate over that data: transforming unstructured data, triggering on fresh data or timers, and automatically retrying steps with clear metrics. This pattern is called "Durable Execution." Cloudflare calls it Workflows.

Workflows run on top of Workers, so the compute is fully managed. A workflow is triggered by an event notification consumed from a Queue — or by an HTTP request, another Worker, or a scheduled timer. Each workflow run defines steps that are individually retriable units of work.

State is durably persisted between steps. Because each step can emit state, any underlying failure or exception can resume execution from the last successful step. Every step call automatically emits metrics tied to the workflow run, providing observability into each unit of execution without extra instrumentation.

An early example uses Workflows to generate text embeddings with Workers AI and stores them in Vectorize as content is written to or updated in R2:

The flow looks like this step-by-step:

Which maps to code like this with the Workflows API:

import { Ai } from "@cloudflare/ai";
import { Workflow } from "cloudflare:workers";

export interface Env {
  R2: R2Bucket;
  AI: any;
  VECTOR_INDEX: VectorizeIndex;
}

export default class extends Workflow {
  async run(event: Event) {
    const ai = new Ai(this.env.AI);

    // List of keys to fetch from our incoming event notification
    const keysToFetch = event.messages.map((val) => {
      return val.object.key;
    });

    // The return value of each step is stored (the "durable" part
    // of "durable execution")
    // This ensures that state can be persisted between steps, reducing
    // the need to recompute results ($$, time) should subsequent
    // steps fail.
    const inputs = await this.ctx.run(
      // Each step has a user-defined label
      // Metrics are emitted as each step runs (to success or failure)
// with this label attached and available within per-Workflow
// analytics in near-real-time.
"read objects from R2", async () => {
      const objects = [];

      for (const key of keysToFetch) {
        const object = await this.env.R2.get(key);
        objects.push(await object.text());
      }

      return objects;
    });

    // Persist the output of this step.
    const embeddings = await this.ctx.run(
      "generate embeddings",
      async () => {
        const { data } = await ai.run("@cf/baai/bge-small-en-v1.5", {
          text: inputs,
        });

        if (data.length) {
          return data;
        } else {
          // Uncaught exceptions trigger an automatic retry of the step
          // Retries and timeouts have sane defaults and can be overridden
    // per step
          throw new Error("Failed to generate embeddings");
        }
      },
      {
        retries: {
          limit: 5,
          delayMs: 1000,
          backoff: "exponential",
        },
      }
    );

    await this.ctx.run("insert vectors", async () => {
      const vectors = [];

      keysToFetch.forEach((key, index) => {
        vectors.push({
          id: crypto.randomUUID(),
          // Our embeddings from the previous step
          values: embeddings[index].values, 
          // The path to each R2 object to map back to during
 	    // vector search
          metadata: { r2Path: key },
        });
      });

      return this.env.VECTOR_INDEX.upsert(vectors);
    });
  }
}

This execution model applies to a wide range of domains, including:

• Deploying software: build and health-check steps gate progress until your deployment meets criteria.
• Post-processing user data: a workflow triggered by an R2 upload parses the data asynchronously, redacts sensitive information, writes sanitized output, and sends a notification.
• Payment and batch workflows: aggregating customer usage data on a schedule, triggering spend alerts, or generating PDF invoices.

More details on Workflows will be shared during the second quarter of 2024, including idempotency, observability, local development, and templates.

Putting the Pieces Together

A full data platform needs to enable three things:

1. Ingesting data in the right format from the right sources
2. Storing that data securely and durably
3. Querying the data to extract insights or transform it for other tools

R2 addressed storage. Pipelines, Event Notifications, and Workflows address the surrounding workflows. The result is an architecture where Pipelines (1) scales out to ingest and batch data, durably stores it in R2 (2), and then makes it available for querying as soon as an Event Notification fires when the data is written. Workflows, ClickHouse, or any query engine can process it without polling.

"Ready" is automatically triggered by the Event Notification, not by external polling. Data doesn't need post-hoc batching or filtering, and query engines don't slow down on unprocessed data. There is also no need to manage infrastructure for load or data jurisdiction requirements.

Cloudflare intends to keep its data products subject to zero egress fees, matching its R2 approach and extending it to Pipelines and other offerings. Feedback is being taken through the Developer Discord and other channels as these services approach their respective public betas.