カフカ、Go、PostgresとGraphicsを使用してページビューの解析システムを作成する

なぜ我々はカフカが必要ですか？

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

カフカを使用する利点

ステップ1：セットアップカフカサーバー

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

ステップ2：ブートストラッププロジェクト

ステップ3：作成プロデューサー

GQLGenを使用したGraphSQLサーバーの作成

Note: if you get validation failed error install the dependencies mentioned in the error: example go get github.com/vektah/gqlparser/[email protected]

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

echo "" > graph/schema.resolvers.go

go run github.com/99designs/gqlgen generate

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
}

query {
  ping {
    message
  }
}

セットアップ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

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

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
}

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

ステップ4：作成消費者

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

セットアップをビルド＆プロデューサーとイベントスキーマ

Note: make sure you are in consumer dir: cd analytics-system/consumer

go get -u gorm.io/gorm
go get -u gorm.io/driver/postgres 

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

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

ステップ5：テストフロー

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