Apache Kafka is an open source distributed event streaming platform used for high performance data pipelines. It can be used for real-time / batch data processing. Typical kafka system looks like
Why we need kafka?
Note: Skip this section if you already know what kafka is and benefits of using kafka😅
Think differently about how data is managed and how extract, transform and load (ETL) technologies are used to.
Earlier, we use to have operational database and on regular intervals we had to transform the data and load into a data warehouse where we can further use it.
But now databases are augmented / replaced by distributed data systems, where we have multiple databases / datasources like Mongodb, Casandradb, Hadoop, etc. to store the data based on the requirements of each system.
ETL tools will have to handle more than databases and data warehouses in case of distributed systems. ETL tools were build to process data in batch fashion. They are resource intensive and time taking processes.
With this new era, applications not only collects the operational data but there are lot of meta data like logs, analytics collected by each of the system.
Also the rise of stream data is increasing where we need to process data on the go instead of processing it in batches.
With this new world of data streaming, we need to ability to process high volume and highly diverse data. Data usually flows in form of events. Consider we have event center which gathers events from different sources and shares with various data sources
Kafka plays this role of Event Center, where data is queued and stored till consumed by consumer.
Benefits of using Kafka
- In case of consumer failures data can be re-gained
- Reduce the cost of ETL as now consumer itself can decided how to use this data
- Asynchronously stream the data
- Can process high volume and diverse data while streaming itself.
For more information you can check this amazing talk by Neha Narkhede on how to think about the data while designing large scale application and how to use Kafka.
Alright lets start building our Analytics System using Kafka. To simplify the example, we will be recording page events from the website and save them to Postgres db. Our system design will look like
Step 1: Setup Kafka Server
For this demo we will be using docker to run kafka server. But for production you can use Confluent or any other hosted service.
- Create
analytics-system/docker-compose.yaml
- Paste the following content in
docker-compose.yaml
version: "3"
services:
zookeeper:
image: wurstmeister/zookeeper
container_name: zookeeper
ports:
- 2181:2181
kafka:
image: wurstmeister/kafka
container_name: kafka
ports:
- 9092:9092
environment:
KAFKA_ADVERTISED_HOST_NAME: localhost
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
ZooKeeper is a centralized service for maintaining configuration information, naming, providing distributed synchronization, and providing group services.
- Start kafka server:
docker-compose up
Step 2: Bootstrap Project
- Create repo
mkdir analytics-system
- Change working dir
cd analytics-system
- Create producer dir
mkdir producer && cd producer
- Init producer project
go mod init producer
- Create consumer dir
cd .. && mkdir consumer && cd consumer
- Init consumer project
go mod init consumer
Step 3: Create Producer
Create Graphql Server using gqlgen
- Change to producer dir
cd analytics-system/producer
- Download the dependency:
go get github.com/99designs/gqlgen
- Initialise Project:
go run github.com/99designs/gqlgen init
Note: if you get validation failed error install the dependencies mentioned in the error: example
go get github.com/vektah/gqlparser/v2@v2.1.0
- Start and test graphql server
go build && ./producer
- Replace the initial boilerplate graphql file with following
analytics-system/producer/graph/schema.graphqls
scalar Int64
type Event {
id: ID!
eventType: String
path: String
search: String
title: String
url: String
userId: String
}
type PingResponse {
message: String!
}
input RegisterKafkaEventInput {
eventType: String!
userId: String!
path: String!
search: String!
title: String!
url: String!
}
type Mutation {
register_kafka_event(event: RegisterKafkaEventInput!): Event!
}
type Query {
ping: PingResponse!
}
Here we are defining mutation and types required to produce page view event.
- Clear content of
analytics-system/producer/graph/schema.resolvers.go
echo "" > graph/schema.resolvers.go
- Generate new resolvers and query as per the graphql file defined above.
go run github.com/99designs/gqlgen generate
- Replace
ping
query resolver to returnHello world
or some string.
Note this step is just to test if our server starts correctly
Update Ping
resolver in graph/schema.resolvers.go
with following content
func (r *queryResolver) Ping(ctx context.Context) (*model.PingResponse, error) {
res := &model.PingResponse{
Message: "Hello world",
}
return res, nil
}
- Build and test server
go build && ./producer
- Hit
http://localhost:8080
in your browser and test the ping query
query {
ping {
message
}
}
Setup Kafka Producer using confluent-kakfka-go
- Install dependencies:
go get -u gopkg.in/confluentinc/confluent-kafka-go.v1/kafka
Note: Latest version of confluent-kafka-go doesn't require
librdkafka
, but in case if you face any errors check the following link and install the require dependencies https://github.com/confluentinc/confluent-kafka-go#installing-librdkafka
- Setup kafka topic:
Add following util in graph/schema.resolvers.go
. This function will make sure that topic is always created
// function to create topic
// sample usage CreateTopic("PAGE_VIEW")
func CreateTopic(topicName string) {
a, err := kafka.NewAdminClient(&kafka.ConfigMap{"bootstrap.servers": "localhost"})
if err != nil {
panic(err)
}
defer a.Close()
maxDur, err := time.ParseDuration("60s")
if err != nil {
panic("ParseDuration(60s)")
}
ctx := context.Background()
results, err := a.CreateTopics(
ctx,
// Multiple topics can be created simultaneously
// by providing more TopicSpecification structs here.
[]kafka.TopicSpecification{{
Topic: topicName,
NumPartitions: 1,
}},
// Admin options
kafka.SetAdminOperationTimeout(maxDur))
if err != nil {
log.Printf("Failed to create topic: %v\n", err)
}
log.Println("results:", results)
}
A Topic is a category/feed name to which records are stored and published
- Produce Kafka Event
Replace RegisterKafkaEvent
resolver function in graph/schema.resolver.go
with following
func (r *mutationResolver) RegisterKafkaEvent(ctx context.Context, event model.RegisterKafkaEventInput) (*model.Event, error) {
p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost"})
if err != nil {
panic(err)
}
defer p.Close()
// Delivery report handler for produced messages
go func() {
for e := range p.Events() {
switch ev := e.(type) {
case *kafka.Message:
if ev.TopicPartition.Error != nil {
fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
} else {
fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
}
}
}
}()
// Produce messages to topic (asynchronously)
topic := event.EventType
CreateTopic(topic)
currentTimeStamp := fmt.Sprintf("%v", time.Now().Unix())
e := model.Event{
ID: currentTimeStamp,
EventType: &event.EventType,
Path: &event.Path,
Search: &event.Search,
Title: &event.Title,
UserID: &event.UserID,
URL: &event.URL,
}
value, err := json.Marshal(e)
if err != nil {
log.Println("=> error converting event object to bytes:", err)
}
p.Produce(&kafka.Message{
TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
Value: []byte(value),
}, nil)
// Wait for message deliveries before shutting down
p.Flush(15 * 1000)
return &e, nil
}
- Test the event produced:
go build && ./producer
- Hit localhost:8080 in browser and test following mutation
mutation {
register_kafka_event(event: {
eventType: "PAGE_VIEW",
userId: "some_session_id",
path: "/test",
search: "?q=foo"
title: "Kafka Demo",
url: "kafka.demo.com"
}) {
id
eventType
}
}
Hurray! 🚀 our producer is ready 🎉
Step 4: Create Consumer
We have already setup consumer project in analytics-system/consumer
. Here in consumer we will listen to events produced by Kafka server in step 3 and save it into postgres db.
Note you can process & transform this data based on system that you want to store into.
To simplify the process we will be using gorm
a SQL ORM(object relational model) for Golang.
Setup gorm
and event schema
Note: make sure you are in consumer dir:
cd analytics-system/consumer
- Install dependencies:
go get -u gorm.io/gorm
go get -u gorm.io/driver/postgres
- Create
main.go
file:touch analytics-system/consumer/main.go
- Connect to db in
main.go
and setup Event Schema
Note: for this example we are using local postgres instance.
analytics-system/consumer/main.go
package main
import (
"log"
"gorm.io/driver/postgres"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"gorm.io/gorm/schema"
)
type Event struct {
ID string `gorm:"primaryKey"`
UserID string
EventType string
Path string
Search string
Title string
URL string
CreatedAt int64 `gorm:"autoCreateTime"` // same as receivedAt
UpdatedAt int64 `gorm:"autoUpdateTime"`
}
func SaveEvent(db *gorm.DB, event Event) (Event, error) {
result := db.Clauses(
clause.OnConflict{
UpdateAll: true,
Columns: []clause.Column{},
}).Create(&event)
if result.Error != nil {
log.Println(result.Error)
return event, result.Error
}
return event, nil
}
func main() {
dbURL :=
`postgres://localhost:5432/postgres`
ormConfig := &gorm.Config{
NamingStrategy: schema.NamingStrategy{
TablePrefix: "kafka_",
},
}
db, err := gorm.Open(postgres.Open(dbURL), ormConfig)
if err != nil {
panic(`Unable to connect to db`)
}
log.Println("=>Connected to successfully:", db)
err = db.AutoMigrate(&Event{})
if err != nil {
log.Println("Error migrating schema:", err)
}
}
Setup kafka Consumer code
Install dependencies:
go get -u gopkg.in/confluentinc/confluent-kafka-go.v1/kafka
Update
main.go
with following content
package main
import (
"encoding/json"
"fmt"
"log"
"github.com/confluentinc/confluent-kafka-go/kafka"
"gorm.io/driver/postgres"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"gorm.io/gorm/schema"
)
type Event struct {
ID string `gorm:"primaryKey"`
UserID string
EventType string
Path string
Search string
Title string
URL string
CreatedAt int64 `gorm:"autoCreateTime"` // same as receivedAt
UpdatedAt int64 `gorm:"autoUpdateTime"`
}
func SaveEvent(db *gorm.DB, event Event) (Event, error) {
result := db.Clauses(
clause.OnConflict{
UpdateAll: true,
Columns: []clause.Column{},
}).Create(&event)
if result.Error != nil {
log.Println(result.Error)
return event, result.Error
}
return event, nil
}
func main() {
dbURL :=
`postgres://localhost:5432/postgres`
ormConfig := &gorm.Config{
NamingStrategy: schema.NamingStrategy{
TablePrefix: "kafka_",
},
}
db, err := gorm.Open(postgres.Open(dbURL), ormConfig)
if err != nil {
panic(`Unable to connect to db`)
}
log.Println("=> Connected to db successfully", db)
err = db.AutoMigrate(&Event{})
if err != nil {
log.Println("Error migrating schema:", err)
}
c, err := kafka.NewConsumer(&kafka.ConfigMap{
"bootstrap.servers": "localhost",
"group.id": "myGroup",
"auto.offset.reset": "earliest",
})
if err != nil {
panic(err)
}
c.SubscribeTopics([]string{"PAGE_VIEW"}, nil)
for {
msg, err := c.ReadMessage(-1)
if err == nil {
fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
var event Event
err := json.Unmarshal(msg.Value, &event)
if err != nil {
log.Println("=> error converting event object:", err)
}
_, err = SaveEvent(db, event)
if err != nil {
log.Println("=> error saving event to db...")
}
} else {
// The client will automatically try to recover from all errors.
fmt.Printf("Consumer error: %v (%v)\n", err, msg)
}
}
}
- Test consumer:
go build && ./consumer
Step 5: Test the flow
Start kafka server if you don't have it running:
docker-compose up
Start producer
cd analytics-system/producer && go build && ./producer
Start consumer
cd analytics-system/consumer && go build && ./consumer
Hit http://locahost:8080 in the browser
Trigger the mutation
mutation {
register_kafka_event(event: {
eventType: "PAGE_VIEW",
userId: "some_session_id",
path: "/test",
search: "?q=foo"
title: "Kafka Demo",
url: "kafka.demo.com"
}) {
id
eventType
}
}
Check the consumer log. You should be able to see logs for data being saved in postgres.
Check the postgres data
SELECT * FROM kafka_events;
Hurray! 🚀 thats all our page view analytics event system is ready. 👏
You can check the full code base on github
Top comments (0)