clickhouse

README

go-clickhouse

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())
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())
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 {
        //
    }
}

External data for query processing

See documentation for details

conn := clickhouse.NewConn("localhost:8123", clickhouse.NewHttpTransport())
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()
conn1 := clickhouse.NewConn("host1", http)
conn2 := clickhouse.NewConn("host2", http)

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

Transport options

Timeout

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

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

Documentation

Index

• type Array
• type Cluster
  • func NewCluster(conn ...*Conn) *Cluster
  • func (c *Cluster) ActiveConn() *Conn
  • func (c *Cluster) Check()
  • func (c *Cluster) IsDown() bool
  • func (c *Cluster) OnCheckError(f PingErrorFunc)
• type Column
• type Columns
• type Conn
  • func NewConn(host string, t Transport) *Conn
  • func (c *Conn) Ping() (err error)
• type DbError
  • func (e *DbError) Code() int
  • func (e *DbError) Error() string
  • func (e *DbError) Message() string
  • func (e *DbError) Response() string
  • func (e *DbError) String() string
• type External
• type Func
• type HttpTransport
  • func NewHttpTransport() HttpTransport
  • func (t HttpTransport) Exec(conn *Conn, q Query, readOnly bool) (res string, err error)
• type Iter
  • func (r *Iter) Error() error
  • func (r *Iter) Scan(vars ...interface{}) bool
• type PingErrorFunc
• type Query
  • func BuildInsert(tbl string, cols Columns, row Row) (Query, error)
  • func BuildMultiInsert(tbl string, cols Columns, rows Rows) (Query, error)
  • func NewQuery(stmt string, args ...interface{}) Query
  • func (q *Query) AddExternal(name string, structure string, data []byte)
  • func (q Query) Exec(conn *Conn) (err error)
  • func (q Query) Iter(conn *Conn) *Iter
• type Row
• type Rows
• type Transport

Types

type Array

type Array []interface{}

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

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

type Func

type Func struct {
	Name string
	Args interface{}
}

type HttpTransport

type HttpTransport struct {
	Timeout time.Duration
}

func NewHttpTransport

func NewHttpTransport() HttpTransport

func (HttpTransport) Exec

func (t HttpTransport) Exec(conn *Conn, q Query, readOnly bool) (res string, err error)

type Iter

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

func (*Iter) Error

func (r *Iter) Error() error

func (*Iter) Scan

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

type PingErrorFunc

type PingErrorFunc func(*Conn)

type Query

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

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

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) (res string, err error)
}