web3-flight-rpc-server

Blockchain data streaming in Apache Arrow format. Web3 data for data engineers, ready for large-scale analysis

Project Overview

This is an implementation of Apache Arrow Flight RPC server for blockchain data. Apache Arrow format allows zero-copy streaming data processing on top of standard eth-web3 RPC.

There are a few key points why to consider flight rpc:

• Plenty of clients implementations including JavaScript, Go, Python, R and Rust (see: https://arrow.apache.org/docs/implementations.html)

• Rich libraries support. Process eth json-rpc data with your favorite data framework:

  • pandas
  • polars
  • apache datafusion
  • and many more natively supporting apache arrow format, all with zero copy and without time loss on serialization and deserialization between the formats

• ADBC Support Ready-made connectors to most popular databases and formats allow you to write data effortlessly

Supported Datasets

All datasets requests supports following params:

• start_block block number to start streaming from (default LATEST)
• end_block block number to stop streaming (default NULL)
• batch_size

Blocks

See schema for stream return type

Schema

private static final Schema BLOCK_SCHEMA = new Schema(List.of(
        Field.nullable(BLOCK_NUMBER, new ArrowType.Int(64, true)),
        Field.nullable(BLOCK_HASH_FIELD, new ArrowType.Utf8()),
        Field.nullable(BLOCK_PARENT_HASH, new ArrowType.Utf8()),
        Field.nullable(BLOCK_TIMESTAMP, new ArrowType.Int(64, true)),
        Field.nullable(BLOCK_AUTHOR, new ArrowType.Utf8()),
        Field.nullable(BLOCK_MINER, new ArrowType.Utf8()),
        Field.nullable(BLOCK_TRANSACTIONS_ROOT, new ArrowType.Utf8()),
        Field.nullable(BLOCK_STATE_ROOT, new ArrowType.Utf8()),
        Field.nullable(BLOCK_RECEIPTS_ROOT, new ArrowType.Utf8()),
        Field.nullable(BLOCK_LOGS_BLOOM, new ArrowType.Utf8()),
        Field.nullable(BLOCK_GAS_LIMIT, new ArrowType.Int(64, true)),
        Field.nullable(BLOCK_GAS_USED, new ArrowType.Int(64, true)),
        Field.nullable(BLOCK_SIZE, new ArrowType.Int(64, true)),
        Field.nullable(BLOCK_EXTRA_DATA, new ArrowType.Utf8()),
        Field.nullable(BLOCK_NONCE, new ArrowType.Utf8()),
        Field.nullable(BLOCK_DIFFICULTY, new ArrowType.Utf8()),
        Field.nullable(BLOCK_TOTAL_DIFFICULTY, new ArrowType.Utf8()),
        Field.nullable(BLOCK_MIX_HASH, new ArrowType.Utf8()),
        Field.nullable(BLOCK_SHA3_UNCLES, new ArrowType.Utf8()),
        new Field(BLOCK_TRANSACTIONS, new FieldType(true, new ArrowType.List(), null),
                Lists.newArrayList(Field.nullable(BLOCK_TX_CHILD_NAME, new ArrowType.Utf8()))),
        new Field(BLOCK_UNCLES, new FieldType(true, new ArrowType.List(), null),
                Lists.newArrayList(Field.nullable(BLOCK_UNCLE_CHILD_NAME, new ArrowType.Utf8()))),
        new Field(BLOCK_SEAL_FIELDS, new FieldType(true, new ArrowType.List(), null),
                Lists.newArrayList(Field.nullable(BLOCK_SEAL_FIELD_CHILD_NAME, new ArrowType.Utf8())))
));

Logs

Additional parameters:

• topics - list of topics to pass to web3 rpc filter
• addresses - list of addresses to pass to web3 rpc filter

Schema

private static final Schema LOG_SCHEMA = new Schema(List.of(
        Field.nullable(LOG_ADDRESS, new ArrowType.Utf8()),
        Field.nullable(LOG_DATA, new ArrowType.Utf8()),
        new Field(LOG_TOPICS, new FieldType(true, new ArrowType.List(), null),
                Lists.newArrayList(Field.nullable(LOG_TOPIC_CHILD_NAME, new ArrowType.Utf8()))),
        Field.nullable(LOG_BLOCK_NUMBER, new ArrowType.Int(64, true)),
        Field.nullable(LOG_TRANSACTION_HASH, new ArrowType.Utf8()),
        Field.nullable(LOG_TRANSACTION_INDEX, new ArrowType.Int(32, true)),
        Field.nullable(LOG_BLOCK_HASH, new ArrowType.Utf8()),
        Field.nullable(LOG_INDEX, new ArrowType.Int(32, true)),
        Field.nullable(LOG_REMOVED, new ArrowType.Bool())
));

Requirements

• Java 21
• Access to an Ethereum node (for the server). Infura was used for development.

Quick Start with Docker

The easiest way to try the project is using Docker Compose with a local Hardhat node:

# Start Hardhat node and Flight server
docker-compose up -d

# Generate test data (optional - creates 50 transactions with logs)
docker-compose --profile test-data up test-data-generator

# View server logs
docker-compose logs -f flight-server

# Stop all services
docker-compose down

The Flight server will be available at localhost:8815, and Hardhat node at localhost:8545.

Python Data Analysis with Jupyter

Launch Jupyter notebook for interactive data analysis:

# Start Jupyter along with other services
docker-compose --profile jupyter up -d

# View Jupyter logs to confirm startup
docker-compose logs jupyter

# Access Jupyter at: http://localhost:8888
# Token: ethereum

Open the notebook ethereum_data_analysis.ipynb to explore:

• Querying blockchain data via Arrow Flight
• Zero-copy conversion to Pandas
• Visualizing logs and blocks
• Analyzing gas usage, events, and transactions
• Filtering by contract address

Note: Generate test data first for meaningful analysis:

docker-compose --profile test-data up test-data-generator

Building the Project

To build the project, run:

mvn clean package

This will create the following JAR files:

• server/target/server.jar - long-running Flight RPC server
• backfill/target/backfill.jar - one-shot batch S3 backfill job
• client/target/client.jar - example client

Running the Applications

Important: VM Options

Application requires the following VM options:

--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED

Running the Server

The server requires an Ethereum node URL and can optionally specify a port for the Flight server.

Using environment variables:

java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar server/target/server.jar

With environment variables:

• HTTP_NODE_URL - URL of the Ethereum node (required)
• WEBSOCKET_NODE_URL - URL of the Ethereum node websocket (required)
• FLIGHT_PORT - Port for the Flight server (default: 8815)

For the full env var reference covering S3 cold-archive, retention, restart-warm cache, and the batch backfill job — and how those settings interact between the server and the backfill — see configuration.md.

The server exposes Prometheus metrics on METRICS_PORT (default 9091) at /metrics. For the full metric reference (ingestion, hot cache, S3 cold tier, subscriptions) and example PromQL alerts, see monitoring.md.

Using command line arguments:

java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar server/target/server.jar <ethereum_node_url> <flight_port>

Example:

java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar server/target/server.jar https://mainnet.infura.io/v3/your-api-key 8815

Running the Client

The client can connect to a Flight server at a specified host and port.

java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar client/target/client.jar [host] [port]

By default, it connects to 127.0.0.1:8815 if no arguments are provided.

Example:

java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar client/target/client.jar 127.0.0.1 8815

Example Workflow

1. Start the server with an Ethereum node URL:

   java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar server/target/server.jar https://mainnet.infura.io/v3/your-api-key 8815

2. In another terminal, start the client:

   java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar client/target/client.jar 127.0.0.1 8815

3. The client will connect to the server and start receiving Ethereum logs data and displaying it in TSV format.

Notes

• The server requires a valid Ethereum node URL to function properly
• For production use with Infura/Alchemy, set WEBSOCKET_NODE_URL and HTTP_NODE_URL environment variables
• The Jupyter notebook provides a complete Python example for data analysis

Python Client Example

See notebooks/ethereum_data_analysis.ipynb for a comprehensive example including:

• Connecting to Flight RPC server with pyarrow
• Querying logs and blocks
• Converting to Pandas DataFrames
• Data visualization with matplotlib/seaborn
• Filtering by contract address and topics

More examples (Polars, DataFusion, R) coming soon!