impalathing

README

go-impala is a Go driver for Apache Impala

As far as we know, this is the only pure golang driver for Apache Impala that has TLS and LDAP support.

We at Bipp, want to make large scale data analytics accesible to every business user.

As part of that, we are commited to making this is a production grade driver that be used in serious enterprise scenarios in lieu of the ODBC/JDBC drivers.

Issues and contributions are welcome.

Using

package main

// Simple program to list databases and the tables

import (
	"context"
	"fmt"
	"log"

	impala "github.com/bippio/go-impala"
)

func main() {
	host := "<impala host>"
	port := 21000

	opts := impala.DefaultOptions

	// enable LDAP authentication:
	opts.UseLDAP = true
	opts.Username = "<ldap username>"
	opts.Password = "<ldap password>"

	// enable TLS
	opts.UseTLS = true
	opts.CACertPath = "/path/to/cacert"

	con, err := impala.Connect(host, port, &opts)
	if err != nil {
		log.Fatal(err)
	}

	ctx := context.Background()

	var rows impala.RowSet

	// get all databases for the connection object
	query := fmt.Sprintf("SHOW DATABASES")
	rows, err = con.Query(ctx, query)
	if err != nil {
		log.Fatal("error in retriving databases: ", err)
	}

	databases := make([]string, 0) // databases will contain all the DBs to enumerate later
	for {
		row := make(map[string]interface{})
		err = rows.MapScan(row)
		if err != nil {
			log.Println(err)
			continue
		}
		if db, ok := row["name"].(string); ok {
			databases = append(databases, db)
		}
	}
	log.Println("List of Databases", databases)

	for _, d := range databases {
		q := "SHOW TABLES IN " + d

		rows, err = con.Query(ctx, q)
		if err != nil {
			log.Printf("error in querying database %s: %s", d, err.Error())
			continue
		}

		tables := make([]string, 0) // databases will contain all the DBs to enumerate later
		for rows.Next(ctx) {
			row := make(map[string]interface{})
			err = rows.MapScan(row)
			if err != nil {
				log.Println(err)
				continue
			}
			if tab, ok := row["name"].(string); ok {
				tables = append(tables, tab)
			}
		}
		log.Printf("List of Tables in Database %s: %v\n", d, tables)
	}
}

Initial fork from impalathing

Documentation

Index

• Variables
• type ColumnSchema
• type Connection
  • func Connect(host string, port int, options *Options) (*Connection, error)
  • func (c *Connection) Close(ctx context.Context) error
  • func (c *Connection) Query(ctx context.Context, query string) (RowSet, error)
• type Options
• type RowSet
• type Status
  • func (s *Status) IsComplete() bool
  • func (s *Status) IsSuccess() bool

Variables

var (
	DefaultOptions = Options{PollIntervalSeconds: 0.1, BatchSize: 1024, BufferSize: 4096}
)

Types

type ColumnSchema

type ColumnSchema struct {
	Name string
	Type string
}

type Connection

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

func Connect

func Connect(host string, port int, options *Options) (*Connection, error)

func (*Connection) Close

func (c *Connection) Close(ctx context.Context) error

func (*Connection) Query

func (c *Connection) Query(ctx context.Context, query string) (RowSet, error)

type Options

type Options struct {
	UseTLS              bool
	CACertPath          string
	UseLDAP             bool
	Username            string
	Password            string
	PollIntervalSeconds float64
	BatchSize           int
	BufferSize          int
}

type RowSet

type RowSet interface {
	Schema(ctx context.Context) []*ColumnSchema
	Next(ctx context.Context) bool
	Err() error
	Scan(dest ...interface{}) error
	Poll(ctx context.Context) (*Status, error)
	Wait(ctx context.Context) (*Status, error)
	FetchAll(ctx context.Context) []map[string]interface{}
	MapScan(dest map[string]interface{}) error
}

A RowSet represents an asyncronous hive operation. You can Reattach to a previously submitted hive operation if you have a valid thrift client, and the serialized Handle() from the prior operation.

type Status

type Status struct {
	Error error
	// contains filtered or unexported fields
}

Represents job status, including success state and time the status was updated.

func (*Status) IsComplete

func (s *Status) IsComplete() bool

func (*Status) IsSuccess

func (s *Status) IsSuccess() bool