Create page view analytics system using Kafka, Go, Postgres & GraphQL in 5 steps

Why we need kafka?

Note: Skip this section if you already know what kafka is and benefits of using kafka😅

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.

Step 1: Setup Kafka Server

• 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

• 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

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!
}

• 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 return Hello world or some string.

Note this step is just to test if our server starts correctly

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:

// 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

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
  }
}

Step 4: Create Consumer

Note you can process & transform this data based on system that you want to store into.

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.

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;