Examples

Use these configs in the Beam pipeline builder or with the API.

JavaScript transforms filter by returning null for records you don’t want. Use lowercase addresses and hashes: Allium data stores them in lowercase, and matching is exact.

Filter logs by contract address

Monitor USDC transfer logs on Polygon by filtering on the USDC contract address.

USDC Transfer Monitor pipeline: Polygon logs, a transform, and a Kafka sink named usdc-logs

{
  "name": "USDC Transfer Monitor",
  "description": "Monitor USDC transfers on Polygon",
  "pipeline_config": {
    "source": {
      "type": "pubsub",
      "chain": "polygon",
      "entity": "log"
    },
    "transforms": [
      {
        "type": "v8",
        "script": "function transform(record) { return record.address === '0x3c499c542cef5e3811e1192ce70d8cc03d5c3359' ? record : null; }"
      }
    ],
    "sinks": [
      {
        "type": "kafka",
        "name": "usdc-logs"
      }
    ]
  }
}

The script matches on the address field, the contract that emitted the log, and drops every other record.

Filter and enrich in one transform

Keep only DEX swap events by event signature, then enrich each record.

{
  "name": "DEX Trade Parser",
  "description": "Parse DEX swap events and extract trade details",
  "pipeline_config": {
    "source": {
      "type": "pubsub",
      "chain": "polygon",
      "entity": "log"
    },
    "transforms": [
      {
        "type": "v8",
        "script": "function transform(record) { if (record.topic0 !== '0xd78ad95fa46c994b6551d0da85fc275fe613ce37657fb8d5e3d130840159d822') return null; record.parsed = true; return record; }"
      }
    ],
    "sinks": [
      {
        "type": "kafka",
        "name": "dex-trades"
      }
    ]
  }
}

topic0 is the event signature hash; this one is the Uniswap V2 swap. Return early on records you don’t need so the rest of the script only runs on matches.

ERC-20 token transfers on Base

Track USDC transfers on Base with the erc20_token_transfer entity instead of raw logs.

{
  "name": "Base Token Transfer Tracker",
  "description": "Monitor ERC-20 token transfers on Base",
  "pipeline_config": {
    "source": {
      "type": "pubsub",
      "chain": "base",
      "entity": "erc20_token_transfer"
    },
    "transforms": [
      {
        "type": "v8",
        "script": "function transform(record) { return record.token_address === '0x833589fcd6edb6e08f4c7c32d4f71b54bda02913' ? record : null; }"
      }
    ],
    "sinks": [
      {
        "type": "kafka",
        "name": "base-usdc-transfers"
      }
    ]
  }
}

The script matches on the token_address field, the token contract. The erc20_token_transfer entity is already decoded (sender, recipient, amount), so you don’t need to parse raw logs.

Filter on a list with a dataset

Monitor any number of contracts in one pipeline by checking each record against a dataset.

{
  "name": "Multi-Token Monitor",
  "description": "Monitor tracked stablecoins on Polygon",
  "pipeline_config": {
    "source": {
      "type": "pubsub",
      "chain": "polygon",
      "entity": "log"
    },
    "transforms": [
      {
        "type": "v8",
        "script": "function transform(record) { return beam.contains('tracked-stablecoins', [record.address]) ? record : null; }"
      }
    ],
    "sinks": [
      {
        "type": "kafka",
        "name": "stablecoin-logs"
      }
    ]
  }
}

Create a dataset named tracked-stablecoins and add 0x3c499c542cef5e3811e1192ce70d8cc03d5c3359 (USDC) and 0xc2132d05d31c914a87c6611c10748aeb04b58e8f (USDT). Add or remove addresses at any time; running pipelines pick up changes with no redeploy.

External Kafka sink

Deliver records to your own Kafka cluster, authenticating with organization secrets.

{
  "name": "External Kafka DEX Monitor",
  "description": "Stream DEX events to an external Kafka cluster",
  "pipeline_config": {
    "source": {
      "type": "pubsub",
      "chain": "ethereum",
      "entity": "log"
    },
    "transforms": [
      {
        "type": "v8",
        "script": "function transform(record) { return record.topic0 === '0xd78ad95fa46c994b6551d0da85fc275fe613ce37657fb8d5e3d130840159d822' ? record : null; }"
      }
    ],
    "sinks": [
      {
        "type": "external_kafka",
        "name": "dex-events",
        "bootstrap_servers": "broker1.example.com:9092,broker2.example.com:9092",
        "username_secret_id": "my-kafka-username",
        "password_secret_id": "my-kafka-password"
      }
    ]
  }
}

Before you use an external_kafka sink, create the username and password secrets in Settings → Secrets. See the configuration reference for all fields.

Queries

Dashboards

Your Data

Beam

Issuer Intelligence

Settings

Filter logs by contract address

Filter and enrich in one transform

ERC-20 token transfers on Base

Filter on a list with a dataset

External Kafka sink