Examples

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

Filter values are managed separately from the pipeline config. After you create a pipeline, add values to its set filters in the UI or with the filter values API. The examples below show only the pipeline config.

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 set filter on address, 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": "redis_set_filter",
        "filter_expr": "root = this.address"
      }
    ],
    "sinks": [
      {
        "type": "kafka",
        "name": "usdc-logs"
      }
    ]
  }
}

root = this.address matches on the address field, the contract that emitted the log. After you create the pipeline, add the USDC contract address (0x3c499c542cef5e3811e1192ce70d8cc03d5c3359) to the filter in the UI or with the filter values API.

Filter and JavaScript transform

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

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

The set filter matches on topic0, the event signature hash. After you create the pipeline, add the Uniswap V2 swap signature (0xd78ad95fa46c994b6551d0da85fc275fe613ce37657fb8d5e3d130840159d822) to the filter. Transforms run in order, so records pass through the filter before the JavaScript transform.

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": "redis_set_filter",
        "filter_expr": "root = this.token_address"
      }
    ],
    "sinks": [
      {
        "type": "kafka",
        "name": "base-usdc-transfers"
      }
    ]
  }
}

root = this.token_address matches on the token_address field, the token contract. After you create the pipeline, add the USDC contract address (0x833589fcd6edb6e08f4c7c32d4f71b54bda02913) to the filter. The erc20_token_transfer entity is already decoded (sender, recipient, amount), so you don’t need to parse raw logs.

Multiple contract addresses

Monitor USDC and USDT on Polygon in one pipeline by adding both addresses to the filter.

{
  "name": "Multi-Token Monitor",
  "description": "Monitor USDC and USDT on Polygon",
  "pipeline_config": {
    "source": {
      "type": "pubsub",
      "chain": "polygon",
      "entity": "log"
    },
    "transforms": [
      {
        "type": "redis_set_filter",
        "filter_expr": "root = this.address"
      }
    ],
    "sinks": [
      {
        "type": "kafka",
        "name": "stablecoin-logs"
      }
    ]
  }
}

After you create the pipeline, add both addresses to the filter: 0x3c499c542cef5e3811e1192ce70d8cc03d5c3359 (USDC) and 0xc2132d05d31c914a87c6611c10748aeb04b58e8f (USDT). A record passes if its field matches any value in the set. Sets hold 10M+ values, enough for very large wallet lists.

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": "redis_set_filter",
        "filter_expr": "root = this.topic0"
      }
    ],
    "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 JavaScript transform

ERC-20 token transfers on Base

Multiple contract addresses

External Kafka sink

Queries

Dashboards

Your Data

Beam

Issuer Intelligence

Settings

​Filter logs by contract address

​Filter and JavaScript transform

​ERC-20 token transfers on Base

​Multiple contract addresses

​External Kafka sink

Filter logs by contract address

Filter and JavaScript transform

ERC-20 token transfers on Base

Multiple contract addresses

External Kafka sink