DEV Community: NightGhost

Creating a ClickHouse cluster - Part II: Replication

NightGhost — Wed, 07 Jul 2021 15:16:42 +0000

In the previous article I showed how to run ClickHouse in cluster mode using only sharding. It's enough for load distribution, but we also need to ensure fault tolerance via replication.

ZooKeeper

Apache ZooKeeper is a coordination service which provides distributed synchronization of config information across nodes. It's obligatory to have a cluster or a single node of it (above 3.4.5) if you want to enable replication on your ClickHouse cluster.

It's pretty easy to launch several nodes of ZooKeeper and assemble them into a cluster. Here's the configuration snippet for docker-compose.yml:

  zk-1:
    container_name: zk_1
    image: bitnami/zookeeper:3.6.1
    environment:
      - ZOO_SERVER_ID=1
      - ALLOW_ANONYMOUS_LOGIN=yes
      - ZOO_SERVERS=zk-1:2888:3888,zk-2:2888:3888

  zk-2:
    container_name: zk_2
    image: bitnami/zookeeper:3.6.1
    environment:
      - ZOO_SERVER_ID=2
      - ALLOW_ANONYMOUS_LOGIN=yes
      - ZOO_SERVERS=zk-1:2888:3888,zk-2:2888:3888

And here's my configuration for a single node:

  ch-zookeeper:
    container_name: ch_zookeeper
    image: zookeeper:3.6.1
    ports:
      - 2181:2181

Pay attention that ClickHouse can work with ZooKeeper cluster only if its nodes are based on the Docker image by bitnami.

ClickHouse does about 10 write operations on ZooKeeper for each INSERT query.

Now you need to tell all the ClickHouse nodes about your ZooKeeper cluster. For this, the config file has a zookeeper section:

    <zookeeper>
        <node index="1">
            <host>zk-1</host>
            <port>2181</port>
        </node>
        <node index="2">
            <host>zk-2</host>
            <port>2181</port>
        </node>
    </zookeeper>

Cluster configuration

I will use 1 master with 2 shards, 2 replicas for each shard. Here's the deployments configuration:

  ch-master:
    container_name: ch_master
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
      - ch-sub-1
      - ch-sub-2
      - ch-sub-3
      - ch-sub-4
    ports:
      - 9000:9000
      - 8123:8123
    volumes:
      - type: volume
        source: ch-master-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-master-logs
        target: /var/log/clickhouse-server
      - ./master-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-1:
    container_name: ch_sub_1
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
    ports:
      - 9001:9000
      - 8124:8123
    volumes:
      - type: volume
        source: ch-sub-1-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-1-logs
        target: /var/log/clickhouse-server
      - ./sub-1-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-2:
    container_name: ch_sub_2
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
    ports:
      - 9002:9000
      - 8125:8123
    volumes:
      - type: volume
        source: ch-sub-2-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-2-logs
        target: /var/log/clickhouse-server
      - ./sub-2-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-3:
    container_name: ch_sub_3
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
    ports:
      - 9003:9000
      - 8126:8123
    volumes:
      - type: volume
        source: ch-sub-3-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-3-logs
        target: /var/log/clickhouse-server
      - ./sub-3-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-4:
    container_name: ch_sub_4
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
    ports:
      - 9004:9000
      - 8127:8123
    volumes:
      - type: volume
        source: ch-sub-4-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-4-logs
        target: /var/log/clickhouse-server
      - ./sub-4-config.xml:/etc/clickhouse-server/config.xml

volumes:
  ch-master-data:
  ch-master-logs:
  ch-sub-1-data:
  ch-sub-1-logs:
  ch-sub-2-data:
  ch-sub-2-logs:
  ch-sub-3-data:
  ch-sub-3-logs:
  ch-sub-4-data:
  ch-sub-4-logs:

The remote_servers section in master-config.xml will now look like this:

    <remote_servers>
        <example_cluster>
            <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>ch-sub-1</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>ch-sub-2</host>
                    <port>9000</port>
                </replica>
            </shard>
            <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>ch-sub-3</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>ch-sub-4</host>
                    <port>9000</port>
                </replica>
            </shard>
        </example_cluster>
    </remote_servers>

As you can see, the internal_replication parameter is now specified for each shard. It's set to true so the replicas themselves will send all their data to other replicas of the shard instead of doing it through the master node. If the parameter is set to false, the master node will replicate all the data on it's own which may lead to an inconsistent state when all the replicas contain different data.

As for configuration files for subordinate nodes, a new section must be defined in them in order to enable replication. This section is called macros.

    <macros>
        <shard>01</shard>
        <replica>ch-sub-1</replica>
    </macros>

It contains parameters that will be accessible through SQL queries related to the node. In the above source, shard and replica are parameters defined by the user, and "01" and "ch-sub-1" are their corresponding values. The first parameter is a cluster identifier and the second one is a replica name which coincides with the domain name specified in docker-compose.yml.

After all the config files are set up, we can finally use scripts to create a cluster and run it. When all the database nodes are up and running, we should first execute our Python scripts for subordinate nodes. All of them look like this:

from clickhouse_driver import Client
from datetime import datetime

if __name__ == "__main__":
    client = Client("127.0.0.1", port="9001")

    client.execute("CREATE DATABASE IF NOT EXISTS billing")

    client.execute(r'''CREATE TABLE IF NOT EXISTS billing.transactions(
                      timestamp DateTime,
                      currency String,
                      value Float64)
                      ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/billing.transactions', '{replica}')
                      PARTITION BY currency
                      ORDER BY timestamp''')

As you can see, the subordinate table now uses ReplicatedMergeTree engine. Its constructor takes the path to the table records in ZooKeeper as the first parameter and the replica name as the second parameter. The path to the table in ZooKeeper should be unique. All the parameters in {} are taken from the aforementioned macros section of the replica config file.

When all the subordinate tables are created, it's time to create a master table. There's no difference from the previous case when only sharding was utilized:

from clickhouse_driver import Client
from datetime import datetime

if __name__ == "__main__":
    client = Client("127.0.0.1", port="9000")

    client.execute("CREATE DATABASE IF NOT EXISTS billing")

    client.execute('''CREATE TABLE IF NOT EXISTS billing.transactions(
                      timestamp DateTime,
                      currency String,
                      value Float64)
                      ENGINE = Distributed(example_cluster, billing, transactions, rand())''')

If you set up all the things properly, you will get a working ClickHouse cluster with replication enabled. The shard is alive if at least one of its replicas is up. Table replication strengthens fault tolerance of the cluster.

All the necessary source code is here.

Playing video in a Golang game

NightGhost — Mon, 04 Jan 2021 04:33:16 +0000

Please proceed here for a new version of the article. The method described below doesn't work anymore (and it's not really good either).

In videogames, we often need to show the player beautiful cinematics, either for ending or in the middle of the playthrough. To do it, we need to somehow load video frames and render them on the screen. Here's how to do that using Pixel game library and goav, Golang bindings for FFmpeg.

Initial setup

First we need to create a GLFW window for rendering OpenGL stuff:

package main

import (
    "fmt"

    "github.com/faiface/pixel"
    "github.com/faiface/pixel/pixelgl"
    colors "golang.org/x/image/colornames"
)

const (
    // WindowWidth is the width of the window.
    WindowWidth = 1280
    // WindowHeight is the height of the window.
    WindowHeight = 720
)

func run() {
    // Create a new window.
    cfg := pixelgl.WindowConfig{
        Title:  "Pixel Rocks!",
        Bounds: pixel.R(0, 0,
            float64(WindowWidth), float64(WindowHeight)),
        VSync:  false,
    }
    win, err := pixelgl.NewWindow(cfg)
    handleError(err)

    fps := 0
    perSecond := time.Tick(time.Second)

    for !win.Closed() {
        win.Clear(colors.White)

        win.Update()

        // Show FPS in the window title.
        fps++

        select {
        case <-perSecond:
            win.SetTitle(fmt.Sprintf("%s | FPS: %d",
                cfg.Title, fps))
            fps = 0

        default:
        }
    }
}

func main() {
    pixelgl.Run(run)
}

func handleError(err error) {
    if err != nil {
        panic(err)
    }
}

Obtaining video frames

Now we need to obtain frames from the video stream of the file. goav provides an example code on how to do that. To make it work with Pixel game library, we need to set the frame decoding format to avcodec.AV_PIX_FMT_RGBA which is used by Pixel.

We also need a way to send the decoded frames to the renderer. For this task we will use a thread-safe frame buffer channel. First we should specify its size:

const (
    FrameBufferSize = 1024
)

The greater the buffer size, the faster the frame transfer from the decoder to the renderer. Now to the channel creation:

frameBuffer := make(chan *pixel.PictureData, FrameBufferSize)

When the frame transfer is complete, we need to close the channel, but we also have to make sure the renderer got all the frames from the buffer. So here's the code for closing the frame buffer:

go func() {
    for {
        if len(frameBuffer) <= 0 {
            close(frameBuffer)
            break
        }
    }
}()

The complete code for reading video frames is presented below:

func readVideoFrames(videoPath string) <-chan *pixel.PictureData {
    // Create a frame buffer.
    frameBuffer := make(chan *pixel.PictureData, FrameBufferSize)

    go func() {
        // Open a video file.
        pFormatContext := avformat.AvformatAllocContext()

        if avformat.AvformatOpenInput(&pFormatContext, videoPath, nil, nil) != 0 {
            fmt.Printf("Unable to open file %s\n", videoPath)
            os.Exit(1)
        }

        // Retrieve the stream information.
        if pFormatContext.AvformatFindStreamInfo(nil) < 0 {
            fmt.Println("Couldn't find stream information")
            os.Exit(1)
        }

        // Dump information about the video to stderr.
        pFormatContext.AvDumpFormat(0, videoPath, 0)

        // Find the first video stream
        for i := 0; i < int(pFormatContext.NbStreams()); i++ {
            switch pFormatContext.Streams()[i].
                CodecParameters().AvCodecGetType() {
            case avformat.AVMEDIA_TYPE_VIDEO:

                // Get a pointer to the codec context for the video stream
                pCodecCtxOrig := pFormatContext.Streams()[i].Codec()
                // Find the decoder for the video stream
                pCodec := avcodec.AvcodecFindDecoder(avcodec.
                    CodecId(pCodecCtxOrig.GetCodecId()))

                if pCodec == nil {
                    fmt.Println("Unsupported codec!")
                    os.Exit(1)
                }

                // Copy context
                pCodecCtx := pCodec.AvcodecAllocContext3()

                if pCodecCtx.AvcodecCopyContext((*avcodec.
                    Context)(unsafe.Pointer(pCodecCtxOrig))) != 0 {
                    fmt.Println("Couldn't copy codec context")
                    os.Exit(1)
                }

                // Open codec
                if pCodecCtx.AvcodecOpen2(pCodec, nil) < 0 {
                    fmt.Println("Could not open codec")
                    os.Exit(1)
                }

                // Allocate video frame
                pFrame := avutil.AvFrameAlloc()

                // Allocate an AVFrame structure
                pFrameRGB := avutil.AvFrameAlloc()

                if pFrameRGB == nil {
                    fmt.Println("Unable to allocate RGB Frame")
                    os.Exit(1)
                }

                // Determine required buffer size and allocate buffer
                numBytes := uintptr(avcodec.AvpictureGetSize(
                    avcodec.AV_PIX_FMT_RGBA, pCodecCtx.Width(),
                    pCodecCtx.Height()))
                buffer := avutil.AvMalloc(numBytes)

                // Assign appropriate parts of buffer to image planes in pFrameRGB
                // Note that pFrameRGB is an AVFrame, but AVFrame is a superset
                // of AVPicture
                avp := (*avcodec.Picture)(unsafe.Pointer(pFrameRGB))
                avp.AvpictureFill((*uint8)(buffer),
                    avcodec.AV_PIX_FMT_RGBA, pCodecCtx.Width(), pCodecCtx.Height())

                // initialize SWS context for software scaling
                swsCtx := swscale.SwsGetcontext(
                    pCodecCtx.Width(),
                    pCodecCtx.Height(),
                    (swscale.PixelFormat)(pCodecCtx.PixFmt()),
                    pCodecCtx.Width(),
                    pCodecCtx.Height(),
                    avcodec.AV_PIX_FMT_RGBA,
                    avcodec.SWS_BILINEAR,
                    nil,
                    nil,
                    nil,
                )

                // Read frames and save first five frames to disk
                packet := avcodec.AvPacketAlloc()

                for pFormatContext.AvReadFrame(packet) >= 0 {
                    // Is this a packet from the video stream?
                    if packet.StreamIndex() == i {
                        // Decode video frame
                        response := pCodecCtx.AvcodecSendPacket(packet)

                        if response < 0 {
                            fmt.Printf("Error while sending a packet to the decoder: %s\n",
                                avutil.ErrorFromCode(response))
                        }

                        for response >= 0 {
                            response = pCodecCtx.AvcodecReceiveFrame(
                                (*avcodec.Frame)(unsafe.Pointer(pFrame)))

                            if response == avutil.AvErrorEAGAIN ||
                                response == avutil.AvErrorEOF {
                                break
                            } else if response < 0 {
                                //fmt.Printf("Error while receiving a frame from the decoder: %s\n",
                                //avutil.ErrorFromCode(response))

                                //return
                            }

                            // Convert the image from its native format to RGB
                            swscale.SwsScale2(swsCtx, avutil.Data(pFrame),
                                avutil.Linesize(pFrame), 0, pCodecCtx.Height(),
                                avutil.Data(pFrameRGB), avutil.Linesize(pFrameRGB))

                            // Save the frame to the frame buffer.
                            frame := getFrameRGBA(pFrameRGB,
                                pCodecCtx.Width(), pCodecCtx.Height())
                            frameBuffer <- frame
                        }
                    }

                    // Free the packet that was allocated by av_read_frame
                    packet.AvFreePacket()
                }

                go func() {
                    for {
                        if len(frameBuffer) <= 0 {
                            close(frameBuffer)
                            break
                        }
                    }
                }()

                // Free the RGB image
                avutil.AvFree(buffer)
                avutil.AvFrameFree(pFrameRGB)

                // Free the YUV frame
                avutil.AvFrameFree(pFrame)

                // Close the codecs
                pCodecCtx.AvcodecClose()
                (*avcodec.Context)(unsafe.Pointer(pCodecCtxOrig)).AvcodecClose()

                // Close the video file
                pFormatContext.AvformatCloseInput()

                // Stop after saving frames of first video straem
                break

            default:
                fmt.Println("Didn't find a video stream")
                os.Exit(1)
            }
        }
    }()

    return frameBuffer
}

We allocate a separate goroutine for the frame decoder so it can do its work in its own pace and just put all the results in the frame buffer.

Converting video frames

Now we need to bring the extracted video frame to the form appropriate for Pixel. First we should extract raw RGBA bytes:

func getFrameRGBA(frame *avutil.Frame, width, height int) *pixel.PictureData {
    pix := []byte{}

    for y := 0; y < height; y++ {
        data0 := avutil.Data(frame)[0]
        buf := make([]byte, width*4)
        startPos := uintptr(unsafe.Pointer(data0)) +
            uintptr(y)*uintptr(avutil.Linesize(frame)[0])

        for i := 0; i < width*4; i++ {
            element := *(*uint8)(unsafe.Pointer(startPos + uintptr(i)))
            buf[i] = element
        }

        pix = append(pix, buf...)
    }

    return pixToPictureData(pix, width, height)
}

To create *pixel.PictureData out of these bytes, we'll use this simple code:

func pixToPictureData(pixels []byte, width, height int) *pixel.PictureData {
    picData := pixel.MakePictureData(pixel.
        R(0, 0, float64(width), float64(height)))

    for y := height - 1; y >= 0; y-- {
        for x := 0; x < width; x++ {
            picData.Pix[(height-y-1)*width+x].R = pixels[y*width*4+x*4+0]
            picData.Pix[(height-y-1)*width+x].G = pixels[y*width*4+x*4+1]
            picData.Pix[(height-y-1)*width+x].B = pixels[y*width*4+x*4+2]
            picData.Pix[(height-y-1)*width+x].A = pixels[y*width*4+x*4+3]
        }
    }

    return picData
}

We need to fill the Pix array vice versa because this is how Pixel treats picture data.

Rendering video frames

Now let's create a new animated sprite to output the video frames:

videoSprite := pixel.NewSprite(nil, pixel.Rect{})
videoTransform := pixel.IM.Moved(pixel.V(
    float64(WindowWidth)/2, float64(WindowHeight)/2))
frameBuffer := readVideoFrames(os.Args[1])

The path to the video file is specified as a command line argument.

Then let's start rendering the video frames:

select {
    case frame, ok: = <-frameBuffer:
        if !ok {
            os.Exit(0)
        }

        if frame != nil {
            videoSprite.Set(frame, frame.Rect)
        }

    default:
}

videoSprite.Draw(win, videoTransform)

Here's the result:

This method can be adapted for other Golang game engines like Ebiten if you know the way to convert RGBA bytes to the appropriate form.

Enjoy the full source code.

Creating a ClickHouse cluster - Part I: Sharding

NightGhost — Thu, 31 Dec 2020 03:26:45 +0000

ClickHouse is an open-source column-oriented DBMS developed by Yandex, a Russian IT company. It's good for Big Data, business analytics and time series data. In this article I'll show you how to run ClickHouse in cluster mode.

Prerequisites

For this tutorial we'll need the official docker image for ClickHouse. Of course, Docker and docker-compose must be installed.

Cluster architecture

Any ClickHouse cluster consists of shards. Shard is a group of nodes storing the same data. If you have a table, its data will be distributed across shards.

Shards consist of replicas. Each replica of the shard stores the same data. All of them are synchronized with each other, so if one replica receives data, it broadcasts the data to the others.

Basically, we need database sharding for load distribution and database replication for fault tolerance.

To query sharded and replicated data tables, we need to use master nodes which form the cluster from subordinate nodes. Master nodes have a couple of distinctive features:

they store cluster configuration;
they have tables on Distributed engine which don't store any rows of data and are intended for querying tables on subordinate nodes and gathering data across the cluster.

Cluster configuration

Well, now onto node configuration. Each node in the cluster should have these fields initialized:

tcp_port - port for the clients to dial and establish a new connection (default value is 9000);
http_port - port for communicating with the clients using HTTP (default value is 8123);
listen_host - IP address to listen on (there can be many of them specified in the configuration file);
interserver_http_port - port for replicas to communicate with each other in the cluster.

Also you can assign values to the parameters such as max_concurrent_queries, max_connections and keep_alive_timeout at your own discretion.

And it's better to leave the default values for these parameters:

logger/log - path to the log file, default value is /var/log/clickhouse-server/clickhouse-server.log;
logger/errorlog - path to the error log file, default value is /var/log/clickhouse-server/clickhouse-server.err.log;
path - path to the data directory, default value is /var/lib/clickhouse/;
tmp_path - path to the temporary files directory, default value is /var/lib/clickhouse/tmp/;
users_config - path to the file with users, their access rights and settings profiles, default value is users.xml;
default_profile - profile with default settings, default value is default;
default_database - the database the client will be connected at first after dialing, default value is default.

Of course, master nodes are special and have to store the cluster configuration. Here's the configuration for a cluster with 3 shards and only 1 replica for each shard:

<remote_servers>
    <example_cluster>
        <shard>
            <replica>
                <host>ch-sub-1</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <replica>
                <host>ch-sub-2</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <replica>
                <host>ch-sub-3</host>
                <port>9000</port>
            </replica>
        </shard>
    </example_cluster>
</remote_servers>

In the host field, an actual domain name is specified. example_cluster is the name of the cluster. You can specify as many clusters as you want in the remote_servers section.

Cluster deployment

Now we are ready to launch the system. I will do it using docker-compose:

version: "3.7"

services:

  ch-master:
    container_name: ch_master
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-sub-1
      - ch-sub-2
      - ch-sub-3
    ports:
      - 9000:9000
      - 8123:8123
    volumes:
      - type: volume
        source: ch-master-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-master-logs
        target: /var/log/clickhouse-server
      - ./master-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-1:
    container_name: ch_sub_1
    image: yandex/clickhouse-server:19.14.13.4
    ports:
      - 9001:9000
      - 8124:8123
    volumes:
      - type: volume
        source: ch-sub-1-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-1-logs
        target: /var/log/clickhouse-server
      - ./sub-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-2:
    container_name: ch_sub_2
    image: yandex/clickhouse-server:19.14.13.4
    ports:
      - 9002:9000
      - 8125:8123
    volumes:
      - type: volume
        source: ch-sub-2-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-2-logs
        target: /var/log/clickhouse-server
      - ./sub-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-3:
    container_name: ch_sub_3
    image: yandex/clickhouse-server:19.14.13.4
    ports:
      - 9003:9000
      - 8126:8123
    volumes:
      - type: volume
        source: ch-sub-3-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-3-logs
        target: /var/log/clickhouse-server
      - ./sub-config.xml:/etc/clickhouse-server/config.xml

volumes:
  ch-master-data:
  ch-master-logs:
  ch-sub-1-data:
  ch-sub-1-logs:
  ch-sub-2-data:
  ch-sub-2-logs:
  ch-sub-3-data:
  ch-sub-3-logs:

As you can see, we created volumes for each node to store the node's data and logs. Also we mounted the node configs. In the current example, we don't need a separate configuration file for each subordinate node, so all of them share the same configuration.

Cluster tables

After everything is up and running, it's time to create data tables. For this task I will use Python programming language and clickhouse-driver library. Now onto the first script, create-cluster.py:

from clickhouse_driver import Client

subs = [
    ("127.0.0.1", "9001"),
    ("127.0.0.1", "9002"),
    ("127.0.0.1", "9003")
]
master = ("127.0.0.1", "9000")

if __name__ == "__main__":
    for sub in subs:
        client = Client(sub[0], port=sub[1])

        client.execute("CREATE DATABASE IF NOT EXISTS db")

        client.execute('''CREATE TABLE IF NOT EXISTS db.entries(
                          timestamp DateTime,
                          parameter String,
                          value Float64)
                          ENGINE = MergeTree()
                          PARTITION BY parameter
                          ORDER BY (timestamp, parameter)''')

    client = Client(master[0], port=master[1])

    client.execute("CREATE DATABASE IF NOT EXISTS db")

    client.execute('''CREATE TABLE IF NOT EXISTS db.entries(
                      timestamp DateTime,
                      parameter String,
                      value Float64)
                      ENGINE = Distributed(example_cluster, db, entries, rand())''')

At first we create a data table on each subordinate node:

CREATE TABLE IF NOT EXISTS db.entries(
timestamp DateTime,
parameter String,
value Float64)
ENGINE = MergeTree()
PARTITION BY parameter
ORDER BY (timestamp, parameter)

Ordering increases performance of SELECT and INSERT queries, whereas partitioning is intended for data manipulations (for example, DROP PARTITION part_name). Keep in mind that the limit of partitions per insert block is controlled by max_partitions_per_insert_block configuration parameter. Recommended total number of partitions per table is 1000..10000. If this threshold is exceeded, it will lead to slow server startup and poor SELECT and INSERT performance.

Then we create a Distributed table on the master node:

CREATE TABLE IF NOT EXISTS db.entries(
timestamp DateTime,
parameter String,
value Float64)
ENGINE = Distributed(example_cluster, db, entries, rand())

The first engine parameter is the cluster name, then goes the name of the database, the table name and a sharding key. The sharding key is an expression whose result is used to decide which shard stores the data row depending on the values of the columns. If you specify rand(), the row goes to the random shard. Sharding key is only applicable if you do INSERT operations on the master table (note that the master table itself doesn't store any data, it only aggregates the data from the shards during queries). But we can perform INSERT operations directly on the subordinate nodes:

from clickhouse_driver import Client
from datetime import datetime

client = Client("127.0.0.1", port="9002")

client.execute("INSERT INTO db.entries (timestamp, parameter, value) VALUES", \
    [(datetime.utcnow(), "elasticity", 38.9), (datetime.utcnow(), "gravity", 27.2), \
        (datetime.utcnow(), "density", 19.8)])

You can insert any data you want to any node. Now try to connect to the master node via ClickHouse client:

docker run --network="clustering_default" -it --rm --link ch_master:clickhouse-server yandex/clickhouse-client:19.14.12.2 --host clickhouse-server

When you are in, try to execute the next set of SQL instructions:

USE db
SELECT * FROM entries

If everything has been set up properly, you'll see all the data you sent to each shard. Why not to try everything yourself? Just look at the project source.

Also you are welcome to take a look at part 2.

Dividing a Bezier curve into equal segments

NightGhost — Thu, 24 Sep 2020 09:21:01 +0000

Sometimes in game development, there's a need to equally distribute homogeneous objects along a curved trajectory. I couldn't find any code solution for this, so here's mine.

Dependencies

For this tutorial I'll use Go programming language and Pixel game library. I'll use Bezier curves because they are flexible enough to create any trajectory and manually edit it, though the algorithm described here is applicable for curved trajectories of any kind. Bezier curve can be created using gonum package. So that's the import section:

import (
    "fmt"
    "time"

    "github.com/faiface/pixel"
    "github.com/faiface/pixel/imdraw"
    "github.com/faiface/pixel/pixelgl"
    colors "golang.org/x/image/colornames"
    "gonum.org/v1/plot/plotter"
    "gonum.org/v1/plot/tools/bezier"
    "gonum.org/v1/plot/vg"
)

As you can see, I also use colornames package for predefined colors.

Creating a curve

First we should create a new curve. I will use cubic Bezier curve with 4 control points:

controlPoints := []vg.Point{
        {X: 0.45, Y: 0.328},
        {X: 1.403, Y: 0.12},
        {X: 0.62, Y: 1.255},
        {X: 1.521, Y: 0.593},
}

As you can see, the numbers are pretty small. Don't worry, the curve will be scaled and translated to look good on the screen.

Also there are constants we'll need later:

const (
    screenWidth              = 1280
    screenHeight             = 720
    offsetX          float64 = 400
    offsetY          float64 = 300
    scaleX           float64 = 300
    scaleY           float64 = 300
    numberOfSegments         = 10
    epsilon          float64 = 0.001
    dt               float64 = 0.5
)

To create a new curve out of the control points:

// Form the curve.
curve := bezier.New(controlPoints...)

Now it's time to compute the curve's points. To obtain a single point of Bezier curve, you need to use parameter t, 0 ≤ t ≤ 1. Method (c Curve) Point(t float64) vg.Point returns a point of the curve corresponding to the parameter. For example, if t = 0.5, the method will return the middle point of the curve.

It's better to get as many Bezier curve points as possible. To do this, we'll use step dt. The lesser the step, the more points you'll obtain.

points := make(plotter.XYs, 0)

for t := 0.0; t < 100.0; t += dt {
    point := curve.Point(t / 100.0)

    points = append(points, plotter.XY{
        X: float64(point.X)*scaleX + offsetX,
        Y: float64(point.Y)*scaleY + offsetY})
}

Drawing the curve

To draw the points of the curve, I will create a new window and an IMDraw object:

cfg := pixelgl.WindowConfig{
    Title:  "Bezier curve",
    Bounds: pixel.R(0, 0, screenWidth, screenHeight),
}
win, err := pixelgl.NewWindow(cfg)
handleError(err)

imd := imdraw.New(nil)

Also I like to setup the FPS counter:

fps := 0
perSecond := time.Tick(time.Second)

Now we are ready to enter the application main loop and draw the curve each frame:

for !win.Closed() {
    win.Clear(colors.White)
    imd.Clear()

    // Draw the curve and other things.
    imd.Color = colors.Red

    for _, point := range points {
        imd.Push(gonumToPixel(point))
        imd.Circle(1, 1)
    }

    imd.Draw(win)

    win.Update()

    // Show FPS in the window title.
    fps++

    select {
    case <-perSecond:
        win.SetTitle(fmt.Sprintf("%s | FPS: %d", cfg.Title, fps))
        fps = 0

    default:
    }
}

By the way, everything written above must be placed inside run() function which is called in main():

func main() {
    pixelgl.Run(run)
}

It's required to make sure the main goroutine won't be assigned to another thread.

Now let's see what we got:

As you can see, the distance between any two adjacent points of the curve is not always the same. Moreover, the graph becomes especially less dense closer to the first and the last control points of the curve.

Actually, that's not what we would like to see. We need all the points to be scattered equally along the curve. To reach it, we must introduce an algorithm of dividing a curve into equal segments.

So let's define a new function getSegmentPoints(points plotter.XYs, numberOfSegments int) []pixel.Vec:

Connect the curve points with lines:

    // Create lines out of bezier
    // curve points.
    lines := []pixel.Line{}

    for i := 0; i < len(points)-1; i++ {
        line := pixel.L(gonumToPixel(points[i]),
            gonumToPixel(points[i+1]))

        lines = append(lines, line)
    }

Note: function gonumToPixel(xy plotter.XY) pixel.Vec
transforms a gonum vector into a pixel vector.

Compute the sum length of the lines.

    // Compute the length
    // of the bezier curve
    // interpolated with lines.
    length := 0.0

    for _, line := range lines {
        length += line.Len()
    }

Compute the step which is the length of a single segment.
```
    step := length / float64(numberOfSegments)
```

Well, we reduced it to a task of segmenting a polygonal chain. The more curve points are obtained, the more precise it will be to the segmenting of the original curve.

First we should do some initialization:

    segmentPoints := []pixel.Vec{}
    lastLine := 0
    lastPoint := lines[0].A
    segmentPoints = append(segmentPoints, lastPoint)

So lastPoint is the last point of the last formed segment. lastLine is the index of the line which contains the last point. Now to the loop:

    for i := 0; i < numberOfSegments; i++ {
        subsegments := []pixel.Line{}
        startLine := pixel.L(lastPoint, lines[lastLine].B)

        subsegments = append(subsegments, startLine)
        localLength := startLine.Len()

        for step-localLength > epsilon {
            line := lines[lastLine+1]
            subsegments = append(subsegments, line)

            localLength += line.Len()
            lastLine++
        }

        line := lines[lastLine]

        if localLength > step {
            difference := localLength - step
            t := difference / line.Len()

            lastPoint = pixel.V(t*line.A.X+(1-t)*line.B.X,
            t*line.A.Y+(1-t)*line.B.Y)
        } else {
            lastPoint = line.B
            lastLine++
        }

        segmentPoints = append(segmentPoints, lastPoint)
    }

In this loop, we pick up lines until their sum length exceeds the length of the segment. When it happens, we compute the segmentation point located on the last line using linear interpolation. If it coincides with the end of the line, we increment the last line counter and start a new iteration with the next line.

Ok, now let's divide our curve into 10 equal segments:

Well, that's much better. Thank you for reading. Here's the source code.

Microservices with Go modules

NightGhost — Fri, 08 Nov 2019 21:27:56 +0000

Problem

I've seen a lot of guides on how to use Go modules to manage dependencies in applications. All of them describe a single-module project with the go.mod file in the project root. In this case, all you need to do is just to copy your go.mod and go.sum files to the application image and launch go mod download.
But the most of Go applications consist of several microservices. Let's consider a project with the structure presented below:

.
├── tracker
│   ├── tracking
│   │   └── tracker.go
│   ├── Dockerfile
│   └── main.go
├── data
│   ├── packets.go
│   └── messages.go
├── docker-compose.yml
├── tgram-bot
│   ├── conf
│   │   └── config.yml
│   ├── Dockerfile
│   ├── main.go
│   └── bot
│       ├── loadconf.go
│       └── bot.go
├── util
│   ├── cache.go
│   ├── storage.go
│   └── converters.go
└── web-server
    ├── api
    │   ├── controller.go
    │   ├── adminController.go
    │   └── middlewares.go
    ├── Dockerfile
    └── main.go

On the scheme above tracker, tgram-bot and web-server are actual microservices whereas data and util are packages used by all the microservices. It's so called multi-module single-repository project. That's a pretty common thing in the world of microservice architecture. And we need to somehow manage dependencies of such project.

Considerations

All the people I've seen on the forums and etc. tried to turn each shared package or microservice into a single Go module, which is a bit wrong approach. Go modules just don't work in such manner. Every Go module is a little GOPATH, so you can launch your application without putting it in the actual GOPATH (read this for more detailed explanation). Linker won't find your cross-shared modules unless you put them in their single git repositories. Here is a tutorial on how to do that. If this procedure is done, you will have go.mod file for each microservice or cross-shared package. That's pretty good because each microservice will have its own isolated set of dependencies for downloading them into its Docker image.
But it's not convenient and straightforward approach. Moreover, it violates the Go modules pattern. Each repository should have only one go.mod file which should be placed in the repository root.

Solution

What I suggest you to do is totally opposite to what is described above. We'll have all the dependencies of our application put in the single pool. For that we've got to write this dependencies.Dockerfile:

FROM golang:1.12 AS dep
# Add the module files and download dependencies.
ENV GO111MODULE=on
COPY ./go.mod /go/src/app/go.mod
COPY ./go.sum /go/src/app/go.sum
WORKDIR /go/src/app
RUN go mod download
# Add the shared packages.
COPY ./data /go/src/app/data
COPY ./util /go/src/app/util

You should build the image with the project root as a build context:

docker build -t dependencies -f ./dependencies.Dockerfile .

This image will keep all the dependencies of the project. We'll use it as a base image for each microservice of the application:

FROM dependencies AS builder
# Copy the application source code.
COPY ./web-server /go/src/app/web-server
# Build the application.
RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \
go build -o /go/bin/web-server /go/src/app/web-server/
ENTRYPOINT [ "/go/bin/web-server" ]

FROM alpine:latest
COPY --from=builder /go/bin/web-server /bin/web-server
ENTRYPOINT [ "/bin/web-server" ]

All the other microservices have similar containerization code. Now we have only one image which keeps all our dependencies inside. It doesn't violate the pattern of Go modules and provides us a little optimization: if we had to use the same library in two different microservices, we would have to download this library twice and store it in each of two images. But with the approach of single dependency pool we need to download it only once and store it only in the one shared image. This approach saves us a bit of time and disk space. Besides, we can put everything that's shared by all our microservices in this image.
And the finishing touch - extending our Makefile:

all: build run

build-dependencies:
    docker build -t dependencies -f ./dependencies.Dockerfile .

build: build-dependencies
    docker-compose build

run:
    docker-compose up

This set of commands will check for new dependencies every time we rebuild the application.
Thank you for reading! I hope this article was helpful for you.