Quickstart

Beta access required. Beam is in beta. Contact [email protected] to request access before you begin.

A Beam pipeline that streams Polygon logs through a transform into a Kafka sink named usdc-logs

Create your first pipeline

Create a pipeline

In Allium App, open Build > Beam and choose New. A new pipeline streams Polygon logs through a JavaScript transform into a Kafka sink. Start with Template builds one from a template instead, and can deploy it in the same step.

Pick the source

Choose a chain and entity to stream, for example polygon and log for every Polygon log. Zero-lag streams from the chain tip before finality, for lower latency at the cost of possible reorged data.

Beam source set to the polygon chain and log entity

Filter the records

Edit the transform and return null for records you don’t want. To keep only USDC on Polygon, match on address, the contract that emitted the log:

function transform(record) {
  return record.address === "0x3c499c542cef5e3811e1192ce70d8cc03d5c3359" ? record : null;
}

To filter on a long or changing list, put the values in a dataset and check them with beam.contains().

Name the sink

Name the Kafka sink, for example usdc-logs. The name becomes part of the topic you consume from.

Deploy

Save keeps the pipeline without running it. Deploy provisions your Kafka topic and credentials and starts the pipeline, usually in under 30 seconds.

Consume your data

Once deployed, Get cooking has the Kafka credentials, the topic, and consumer code with the credentials filled in.

Get cooking panel with Kafka credentials and the pipeline's topic

from confluent_kafka import Consumer

conf = {
    "bootstrap.servers": "<bootstrap-server>",
    "group.id": "<group-id-prefix>my-consumer-group",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": "<username>",
    "sasl.password": "<password>",
    "auto.offset.reset": "latest",
}

consumer = Consumer(conf)
consumer.subscribe(["beam.<config_id>.usdc-logs"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    print(msg.value().decode("utf-8"))

import { Kafka } from "kafkajs";

const kafka = new Kafka({
  brokers: ["<bootstrap-server>"],
  ssl: true,
  sasl: {
    mechanism: "scram-sha-512",
    username: "<username>",
    password: "<password>",
  },
});

const consumer = kafka.consumer({ groupId: "<group-id-prefix>my-consumer-group" });

await consumer.connect();
await consumer.subscribe({ topic: "beam.<config_id>.usdc-logs" });

await consumer.run({
  eachMessage: async ({ message }) => {
    console.log(message.value?.toString());
  },
});

Replace the placeholders with the values from Get cooking. Start the consumer group ID with your group ID prefix.

Verify your deployment

A deployed pipeline previews the latest messages from each sink, and reports worker health, events processed, latency, and errors over the last hour, day, or week. Every worker should be healthy. If one isn’t, see Troubleshooting.

Deployment messages preview for a healthy pipeline's Kafka sink

Deployment health with one healthy worker, events processed and latency charts

Updating a pipeline

Edit the source, transforms, or sinks and deploy again. Deploys are idempotent and zero-downtime; changes apply in place.

Don’t tear down before redeploying. Tearing down causes unnecessary downtime.

Next steps

Configuration reference

Sources, transforms, and sinks

Examples

Ready-to-use pipeline configs for common use cases

Queries

Dashboards

Your Data

Beam

Issuer Intelligence

Settings

Create your first pipeline

Verify your deployment

Updating a pipeline

Next steps

Configuration reference

Examples