A Kafka Connector for listening to blockchain events.
KCQL is not supported.
A Kafka Connector to hook into the live streaming providing a real-time feed for new bitcoin blocks and transactions provided by www.blockhain.info The connector subscribes to notification on blocks, transactions or an address and receives JSON objects describing a transaction or block when an event occurs. This JSON is then pushed via Kafka connect to a Kafka topic and therefore can be consumed either by a Sink or a live stream processor.
Since is a direct WebSocket connection the source will only ever use one connector task at any point. There is no point spawning more and then have duplicate data.
One thing to remember is the subscription API from blockchain doesn’t offer an option to start from a given timestamp. This means if the connect worker is down then you will miss some data.
- Copy the docker-compose file.
- Bring up the stack.
export CONNECTOR=blockchain docker-compose up -d fastdata
If you are using Lenses, login into Lenses and navigate to the connectors page , select Blockchain as the source and paste the following:
name=blockchain connector.class=com.datamountaineer.streamreactor.connect.blockchain.source.BlockchainSourceConnector connect.blockchain.source.kafka.topic=blockchain tasks.max=1
To start the connector without using Lenses, log into the fastdatadev container:
docker exec -ti fastdata /bin/bash
and create a connector.properties file containing the properties above.
Create the connector, with the connect-cli :
connect-cli create blockchain < connector.properties
Wait a for the connector to start and check its running:
connect-cli status blockchain
Check the records in Lenses or with via the console:
kafka-avro-console-consumer \ --bootstrap-server localhost:9092 \ --topic blockchain \ --from-beginning
Bring down the stack:
|connect.blockchain.source.url||The websocket connection.||string||wss://ws.blockchain.info/inv|
|connect.blockchain.source.subscription.addresses||Comma separated list of addresses to receive transactions updates for||string|
|connect.blockchain.source.kafka.topic||Specifies the kafka topic to sent the records to.||string|
|connect.progress.enabled||Enables the output for how many records have been processed||boolean||false|