Introduction
In this project, I set out to solve a common challenge in crypto analytics: how do you reliably stream, store, and visualize fast-moving market data in real time?
To tackle this, I built a production-grade real-time data pipeline that extracts data from the Binance API, loads it into a relational database like PostgreSQL, replicates updates into Cassandra using Change Data Capture (CDC) via Debezium, and finally visualizes key metrics using Grafana.
The goal? Create a seamless, scalable pipeline capable of tracking market trends, spotting top-performing tokens, and powering real-time dashboardswith zero manual refreshes.
Why This Project Matters
Cryptocurrency markets operate 24/7 with volatile price action and massive data velocity. Traditional batch pipelines or manual scripts simply can’t keep up. I wanted to build something that could:
- Continuously ingest price, volume, and trade data from Binance
- Store data in a queryable format for structured analytics
- Replicate changes across systems without breaking consistency
- Enable live, real-time dashboards to support fast decisions
This documentation walks through each phase, from API extraction to CDC to visualization and shares lessons learned along the way.
Data Extraction
Here are the five endpoints I used for this project, each chosen to cover a critical slice of the crypto market landscape:
Latest Prices – /api/v3/ticker/price
This endpoint returns the latest market price for every trading pair. It’s lightweight and perfect for fast polling.
24h Ticker Stats – /api/v3/ticker/24hr
Provides daily stats like price change %, volume, and high/low prices, essential for tracking the top gainers and market trends.
fields:
- priceChangePercent
- highPrice
- volume
Order Book – /api/v3/depth
Returns bid/ask levels for a given trading pair, offering insights into market depth and liquidity. Useful for advanced analytics like order flow tracking.
Recent Trades – /api/v3/trades
Gives a rolling list of individual trade executions, including price, quantity, and timestamps — key for volatility and momentum indicators.
Klines (Candlesticks) – /api/v3/klines
Returns historical OHLCV candlestick data for any symbol over a chosen interval (e.g., 1m, 5m, 1h). This powers most visual time-series analysis in Grafana.
Binance Extraction Script Workflow
- Send API request to a specific Binance endpoint (e.g., /ticker/price).
- Parse JSON response and extract key fields like symbol, price, volume, timestamp.
- Format and clean data (add timestamps, remove duplicates).
- Insert into PostgreSQL staging tables with proper indexing.
- Log activity and handle API failures with retries.
Clone the Project Repository
We start by cloning the CDC project files including table schemas, Debezium config, and consumer scripts:
git clone https://github.com/yourusername/crypto-cdc-pipeline.git
cd crypto-cdc-pipeline
Install Java, Kafka, and Zookeeper
Kafka and Debezium require Java, so we install OpenJDK:
sudo apt update
sudo apt install -y default-jdk
Next, download and extract Kafka:
wget https://downloads.apache.org/kafka/3.6.0/kafka_2.13-3.6.0.tgz
tar -xzf kafka_2.13-3.6.0.tgz
mv kafka_2.13-3.6.0 kafka && cd kafka
Start Zookeeper and Kafka in two separate terminal windows or tmux panes:
# Terminal 1
bin/zookeeper-server-start.sh config/zookeeper.properties
# Terminal 2
bin/kafka-server-start.sh config/server.properties
Install & Configure Debezium Kafka Connect
Debezium captures Postgres changes via Kafka Connect. We install it like this:
cd ~/kafka
mkdir plugins && cd plugins
wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.5.1.Final/debezium-connector-postgres-2.5.1.Final-plugin.tar.gz
tar -xvf debezium-connector-postgres-2.5.1.Final-plugin.tar.gz
Update Kafka Connect config to load this plugin:
nano ../config/connect-distributed.properties
# Add or edit:
plugin.path=/home/youruser/kafka/plugins
Start Kafka Connect:
cd ~/kafka
bin/connect-distributed.sh config/connect-distributed.properties
Create Required Internal Kafka Topics
These are used by Kafka Connect to manage connector state:
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic connect-configs --partitions 1 --replication-factor 1 --config cleanup.policy=compact
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic connect-offsets --partitions 1 --replication-factor 1 --config cleanup.policy=compact
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic connect-status --partitions 1 --replication-factor 1 --config cleanup.policy=compact
Prepare PostgreSQL for CDC
In Dbeaver, log in to your database and create a publication:
CREATE PUBLICATION my_publication FOR TABLE binance.binance_latest_prices, binance.binance_24h_stats, binance.binance_klines, binance.binance_order_book, binance.binance_recent_trades;
Make sure your user has replication rights.
Register the Debezium Connector
Create the config file postgres-connector.json:
{
"name": "postgres-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"database.hostname": "<aiven-host>",
"database.port": "25060",
"database.user": "<username>",
"database.password": "<password>",
"database.dbname": "defaultdb",
"topic.prefix": "crypto_pg",
"plugin.name": "pgoutput",
"publication.name": "my_publication",
"slot.name": "cdc_slot",
"table.include.list": "binance.binance_latest_prices,binance.binance_24h_stats,binance.binance_klines,binance.binance_order_book,binance.binance_recent_trades",
"database.sslmode": "require"
}
}
Register the connector:
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
--data @postgres-connector.json
Check status:
curl http://localhost:8083/connectors/postgres-connector/status | jq
Install Cassandra
Cassandra doesn’t support Python 3.12+. You must use Python 3.10 or 3.11.
Check your messages in the topic
bin/kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic <your-topic-name> \
--from-beginning
Install Cassandra:
echo "deb https://debian.cassandra.apache.org 41x main" | sudo tee /etc/apt/sources.list.d/cassandra.sources.list
curl https://downloads.apache.org/cassandra/KEYS | sudo apt-key add -
sudo apt update
sudo apt install cassandra
Start the Cassandra service:
sudo systemctl start cassandra
sudo systemctl status cassandra
Create Cassandra Keyspace & Tables
Enter the CQL shell:
cqlsh
Create a keyspace:
CREATE KEYSPACE crypto_data WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1};
USE crypto_data;
Create each table one by one(cassandra schema in the repo):
Create Python Consumers for CDC
Each script listens to a Kafka topic and writes to a Cassandra table.
Run each script:
python consumer_latest_prices.py
python consumer_24h_stats.py
python consumer_klines.py
python consumer_order_book.py
python consumer_recent_trades.py
Connect Cassandra to Grafana
grafana-cli plugins install hadesarchitect-cassandra-datasource
systemctl restart grafana-server
Add datasource: In Grafana UI, go to Configuration → Data Sources and add “Apache Cassandra” (community).
Configure: Set Contact point to hostname:9042 (e.g. localhost:9042 if on the same VM), fill in Keyspace, User, and Password
Test & save: Click Save & Test. Once “Database Connection OK” appears, you can create dashboards querying Cassandra.
Grafana Dashboard
Check out the repo:
https://github.com/KellyKiprop/Binance-CDC-Project
Top comments (3)
the viz is on point
good work!
Fantastic job