clickhouse

package module
v1.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Feb 17, 2021 License: MIT Imports: 16 Imported by: 0

README

go-clickhouse + copy CSV and transport performance improvements

Build Status Coverage Status Go Report

Golang Yandex ClickHouse connector

ClickHouse manages extremely large volumes of data in a stable and sustainable manner. It currently powers Yandex.Metrica, world’s second largest web analytics platform, with over 13 trillion database records and over 20 billion events a day, generating customized reports on-the-fly, directly from non-aggregated data. This system was successfully implemented at CERN’s LHCb experiment to store and process metadata on 10bn events with over 1000 attributes per event registered in 2011.

Examples

Query rows
conn := clickhouse.NewConn("localhost:8123", clickhouse.NewHttpTransport(32))
query := clickhouse.NewQuery("SELECT name, date FROM clicks")
iter := query.Iter(conn)
var (
    name string
    date string
)
for iter.Scan(&name, &date) {
    //
}
if iter.Error() != nil {
    log.Panicln(iter.Error())
}
Single insert
conn := clickhouse.NewConn("localhost:8123", clickhouse.NewHttpTransport(32))
query, err := clickhouse.BuildInsert("clicks",
    clickhouse.Columns{"name", "date", "sourceip"},
    clickhouse.Row{"Test name", "2016-01-01 21:01:01", clickhouse.Func{"IPv4StringToNum", "192.0.2.192"}},
)
if err == nil {
    err = query.Exec(conn)
    if err == nil {
        //
    }
}
CSV file insert
conn := clickhouse.NewConn("localhost:8123", clickhouse.NewHttpTransport(32))
query, err := clickhouse.BuildCSVInsert("clicks", csvFile)
if err == nil {
    err = query.Exec(conn)
    if err == nil {
        //
    }
}
External data for query processing

See documentation for details

conn := clickhouse.NewConn("localhost:8123", clickhouse.NewHttpTransport(32))
query := clickhouse.NewQuery("SELECT Num, Name FROM extdata")
query.AddExternal("extdata", "Num UInt32, Name String", []byte("1	first\n2	second")) // tab separated


iter := query.Iter(conn)
var (
    num  int
    name string
)
for iter.Scan(&num, &name) {
    //
}
if iter.Error() != nil {
    log.Panicln(iter.Error())
}

Cluster

Cluster is useful if you have several servers with same Distributed table (master). In this case you can send requests to random master to balance load.

  • cluster.Check() pings all connections and filters active ones
  • cluster.ActiveConn() returns random active connection
  • cluster.OnCheckError() is called when any connection fails

Important: You should call method Check() at least once after initialization, but we recommend to call it continuously, so ActiveConn() will always return filtered active connection.

http := clickhouse.NewHttpTransport(32)
conn1 := clickhouse.NewConn("host1", http)
conn2 := clickhouse.NewConn("host2", http)

cluster := clickhouse.NewCluster(conn1, conn2)
cluster.OnCheckError(func (c *clickhouse.Conn, err error) {
    log.Fatalf("Clickhouse connection failed %s %s", c.Host, err)
})
// Ping connections every second
go func() {
    for {
        cluster.Check()
        time.Sleep(time.Second)
    }
}()

Transport options

Timeout
bufferPoolSize := 32
t := clickhouse.NewHttpTransport(bufferPoolSize)
t.Timeout = time.Second * 5

conn := clickhouse.NewConn("host", t)

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Array

type Array []interface{}

type BufferPool added in v1.3.0

type BufferPool struct {
	// contains filtered or unexported fields
}

func NewBufferPool added in v1.3.0

func NewBufferPool(size int) (bp *BufferPool)

func (*BufferPool) Get added in v1.3.0

func (bp *BufferPool) Get() (b *bytes.Buffer)

func (*BufferPool) Put added in v1.3.0

func (bp *BufferPool) Put(b *bytes.Buffer)

type Cluster

type Cluster struct {
	// contains filtered or unexported fields
}

func NewCluster

func NewCluster(conn ...*Conn) *Cluster

func (*Cluster) ActiveConn

func (c *Cluster) ActiveConn() *Conn

func (*Cluster) Check

func (c *Cluster) Check()

func (*Cluster) IsDown

func (c *Cluster) IsDown() bool

func (*Cluster) OnCheckError

func (c *Cluster) OnCheckError(f PingErrorFunc)

type Column

type Column string

type Columns

type Columns []string

type Conn

type Conn struct {
	Host string
	// contains filtered or unexported fields
}

func NewConn

func NewConn(host string, t Transport) *Conn

func (*Conn) Ping

func (c *Conn) Ping() (err error)

type DbError

type DbError struct {
	// contains filtered or unexported fields
}

func (*DbError) Code

func (e *DbError) Code() int

func (*DbError) Error

func (e *DbError) Error() string

func (*DbError) Message

func (e *DbError) Message() string

func (*DbError) Response

func (e *DbError) Response() string

func (*DbError) String

func (e *DbError) String() string

type External added in v1.3.0

type External struct {
	Name      string
	Structure string
	Data      []byte
}

type Func added in v1.3.0

type Func struct {
	Name string
	Args interface{}
}

type HttpTransport

type HttpTransport struct {
	Timeout    time.Duration
	BufferPool *BufferPool
}

func NewHttpTransport

func NewHttpTransport(bpSize int) HttpTransport

func (HttpTransport) Exec

func (t HttpTransport) Exec(conn *Conn, q Query, readOnly bool) (r io.ReadCloser, err error)

type Iter

type Iter struct {
	// contains filtered or unexported fields
}

func (*Iter) Columns added in v1.3.0

func (r *Iter) Columns() []string

func (*Iter) Error

func (r *Iter) Error() error

func (*Iter) Scan

func (r *Iter) Scan(vars ...interface{}) bool

type PingErrorFunc

type PingErrorFunc func(conn *Conn, err error)

type Query

type Query struct {
	Stmt string
	// contains filtered or unexported fields
}

func BuildCSVInsert added in v1.3.0

func BuildCSVInsert(tbl string, body io.Reader) Query

func BuildInsert

func BuildInsert(tbl string, cols Columns, row Row) (Query, error)

func BuildMultiInsert

func BuildMultiInsert(tbl string, cols Columns, rows Rows) (Query, error)

func NewQuery

func NewQuery(stmt string, args ...interface{}) Query

func (*Query) AddExternal added in v1.3.0

func (q *Query) AddExternal(name string, structure string, data []byte)

func (Query) Exec

func (q Query) Exec(conn *Conn) (err error)

func (Query) Iter

func (q Query) Iter(conn *Conn) *Iter

type Row

type Row []interface{}

type Rows

type Rows []Row

type Transport

type Transport interface {
	Exec(conn *Conn, q Query, readOnly bool) (io.ReadCloser, error)
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL