gohive

README

GoHive

GoHive is a driver for Hive and the Spark Distributed SQL Engine in go that supports connection mechanisms KERBEROS(Gssapi Sasl), NONE(Plain Sasl), LDAP, CUSTOM and NOSASL, both for binary and HTTP transport, with and without SSL, compliant with the database/sql interface. The kerberos mechanism will pick a different authentication level depending on hive.server2.thrift.sasl.qop.

GoHive also offers support to query the Hive metastore with various authentication mechanisms, including KERBEROS.

New Features in 2.0

• Full support of the database/sql interface

For detailed migration instructions, please see the Migration Guide.

Installation

GoHive can be installed with:

go get github.com/ichsansaid/gohive/v2

To add kerberos support GoHive requires header files to build against the GSSAPI C library. They can be installed with:

• Ubuntu: sudo apt-get install libkrb5-dev
• MacOS: brew install krb5
• Debian: yum install -y krb5-devel

Then:

go get -tags kerberos github.com/ichsansaid/gohive/v2

Quickstart

Connection to Hive

GoHive supports the standard database/sql interface:

import (
    "database/sql"
    _ "github.com/ichsansaid/gohive"
)

// Open a connection
// Format: hive://username:password@host:port/database
db, err := sql.Open("hive", "hive://username:password@hs2.example.com:10000/default")
if err != nil {
    log.Fatal(err)
}
defer db.Close()

// Execute a query
rows, err := db.Query("SELECT * FROM my_table LIMIT 10")
if err != nil {
    log.Fatal(err)
}
defer rows.Close()

// Process results
for rows.Next() {
    var id int
    var name string
    if err := rows.Scan(&id, &name); err != nil {
        log.Fatal(err)
    }
    fmt.Printf("id: %d, name: %s\n", id, name)
}

Supported connections

Connect with SASL KERBEROS:

import (
    "database/sql"
    _ "github.com/ichsansaid/gohive"
)

db, err := sql.Open("hive", "hive://hs2.example.com:10000/default?auth=KERBEROS&service=hive")
if err != nil {
    log.Fatal(err)
}
defer db.Close()

This implies setting in hive-site.xml:

• hive.server2.authentication = KERBEROS
• hive.server2.authentication.kerberos.principal = hive/_HOST@EXAMPLE.COM
• hive.server2.authentication.kerberos.keytab = path/to/keytab.keytab

Connect using Plain SASL:

import (
    "database/sql"
    _ "github.com/ichsansaid/gohive"
)

db, err := sql.Open("hive", "hive://myUsername:myPassword@hs2.example.com:10000/default?auth=NONE")
if err != nil {
    log.Fatal(err)
}
defer db.Close()

This implies setting in hive-site.xml:

• hive.server2.authentication = NONE

Connect using NOSASL:

import (
    "database/sql"
    _ "github.com/ichsansaid/gohive"
)

db, err := sql.Open("hive", "hive://@hs2.example.com:10000/default?auth=NOSASL")
if err != nil {
    log.Fatal(err)
}
defer db.Close()

This implies setting in hive-site.xml:

• hive.server2.authentication = NOSASL

Connect using HTTP transport mode

Binary transport mode is supported for auth mechanisms PLAIN, KERBEROS and NOSASL. HTTP transport mode is supported for PLAIN and KERBEROS:

import (
    "database/sql"
    _ "github.com/ichsansaid/gohive"
)

db, err := sql.Open("hive", "hive://@hs2.example.com:10000/default?auth=NOSASL&service=hive&transport=http&httpPath=cliservice")
if err != nil {
    log.Fatal(err)
}
defer db.Close()

This implies setting in hive-site.xml:

• hive.server2.authentication = KERBEROS, or NONE
• hive.server2.transport.mode = http
• hive.server2.thrift.http.port = 10001

Programmatic Connection (HiveConnector)

Instead of building a DSN string, you can use HiveConnector with a structured Config. This is useful when integrating with GORM or when connection parameters come from application config.

Basic usage

import gohive "github.com/ichsansaid/gohive/v2"

db := gohive.OpenDB(gohive.Config{
    Host:     "hs2.example.com",
    Port:     10000,
    Auth:     "NONE",
    Username: "hive",
    Password: "hive",
    Database: "default",
})
defer db.Close()

With CA certificate (server verification)

import gohive "github.com/ichsansaid/gohive/v2"

db := gohive.OpenDB(gohive.Config{
    Host:      "hs2.example.com",
    Port:      10000,
    Auth:      "NONE",
    Username:  "hive",
    Password:  "hive",
    Database:  "default",
    SSLCAFile: "/path/to/ca-cert.crt",
})
defer db.Close()

With client certificate and key (mutual TLS)

import gohive "github.com/ichsansaid/gohive/v2"

db := gohive.OpenDB(gohive.Config{
    Host:            "hs2.example.com",
    Port:            10000,
    Auth:            "NONE",
    Username:        "hive",
    Password:        "hive",
    Database:        "default",
    SSLCertFile:     "/path/to/client-cert.pem",
    SSLKeyFile:      "/path/to/client-key.pem",
    SSLInsecureSkip: false,
})
defer db.Close()

With custom TLS config

For full control over TLS settings, provide your own *tls.Config:

import (
    "crypto/tls"
    "crypto/x509"
    "os"

    gohive "github.com/ichsansaid/gohive/v2"
)

caPEM, _ := os.ReadFile("/path/to/ca-cert.crt")
caPool := x509.NewCertPool()
caPool.AppendCertsFromPEM(caPEM)

clientCert, _ := tls.LoadX509KeyPair("/path/to/client-cert.pem", "/path/to/client-key.pem")

db := gohive.OpenDB(gohive.Config{
    Host:     "hs2.example.com",
    Port:     10000,
    Auth:     "NONE",
    Username: "hive",
    Password: "hive",
    Database: "default",
    TLSConfig: &tls.Config{
        RootCAs:            caPool,
        Certificates:       []tls.Certificate{clientCert},
        InsecureSkipVerify: false,
        MinVersion:         tls.VersionTLS12,
    },
})
defer db.Close()

Note: When TLSConfig is provided directly, SSLCertFile, SSLKeyFile, SSLCAFile, and SSLInsecureSkip are ignored.

Connection to the Hive Metastore

The thrift client is directly exposed, so the API exposed by the Hive metastore can be called directly.

    configuration := gohive.NewMetastoreConnectConfiguration()
    connection, err := gohive.ConnectToMetastore("hm.example.com", 9083, "KERBEROS", configuration)
    if err != nil {
        log.Fatal(err)
    }
    database := hive_metastore.Database{
        Name:        "my_new_database",
        LocationUri: "/"}
    err = connection.Client.CreateDatabase(context.Background(), &database)
    if err != nil {
        log.Fatal(err)
    }
    databases, err := connection.Client.GetAllDatabases(context.Background())
    if err != nil {
        log.Fatal(err)
    }
    log.Println("databases ", databases)
    connection.Close()

Running tests

Tests can be run with:

./scripts/integration

This uses dhive and it will start three docker instances with Hive, the Hive metastore, and Kerberos. kinit, klist, kdestroy have to be installed locally. hs2.example.com and hm.example.com will have to be an alias for 127.0.0.1 in /etc/hosts. The krb5 configuration file should be created with bash scripts/create_krbconf.sh. Overall the steps used in the travis CI can be followed.

Documentation

Index

• Constants
• Variables
• func OpenDB(cfg Config) *sql.DB
• type Config
• type DSN
  • func ParseDSN(dsn string) (*DSN, error)
• type Driver
  • func (d *Driver) Open(name string) (driver.Conn, error)
  • func (d *Driver) OpenConnector(name string) (driver.Connector, error)
• type HiveConnector
  • func NewConnector(cfg Config) *HiveConnector
  • func (c *HiveConnector) Connect(ctx context.Context) (driver.Conn, error)
  • func (c *HiveConnector) Driver() driver.Driver
• type HiveMetastoreClient
  • func ConnectToMetastore(host string, port int, auth string, ...) (client *HiveMetastoreClient, err error)
  • func (c *HiveMetastoreClient) Close()
• type MetastoreConnectConfiguration
  • func NewMetastoreConnectConfiguration() *MetastoreConnectConfiguration
• type TSaslTransport
  • func NewTSaslTransport(trans thrift.TTransport, host string, mechanismName string, ...) (*TSaslTransport, error)
  • func (p *TSaslTransport) Close() (err error)
  • func (p *TSaslTransport) Flush(ctx context.Context) (err error)
  • func (p *TSaslTransport) IsOpen() bool
  • func (p *TSaslTransport) Open() (err error)
  • func (p *TSaslTransport) Read(buf []byte) (l int, err error)
  • func (p *TSaslTransport) RemainingBytes() uint64
  • func (p *TSaslTransport) SetMaxLength(maxLength uint32)
  • func (p *TSaslTransport) Write(buf []byte) (int, error)

Constants

const (
	START    = 1
	OK       = 2
	BAD      = 3
	ERROR    = 4
	COMPLETE = 5
)

Variables

var DEFAULT_ERROR_CODE = int32(-1)

var DEFAULT_ERROR_MESSAGE = "unknown error"

var DEFAULT_SQL_STATE = ""

var DEFAULT_STATUS = hiveserver.TStatus{
	StatusCode:   hiveserver.TStatusCode_ERROR_STATUS,
	InfoMessages: nil,
	SqlState:     &DEFAULT_SQL_STATE,
	ErrorCode:    &DEFAULT_ERROR_CODE,
	ErrorMessage: &DEFAULT_ERROR_MESSAGE,
}

Functions

func OpenDB

func OpenDB(cfg Config) *sql.DB

OpenDB is a convenience that returns a *sql.DB ready for use with GORM.

Types

type Config

type Config struct {
	Host              string
	Port              int
	Auth              string // "NONE", "KERBEROS", "NOSASL", etc.
	Username          string
	Password          string
	Database          string
	TransportMode     string // "binary" or "http"
	HTTPPath          string
	Service           string // Kerberos service name
	TLSConfig         *tls.Config
	SSLCertFile       string
	SSLKeyFile        string
	SSLCAFile         string
	SSLInsecureSkip   bool
	HiveConfiguration map[string]string
}

Config holds everything needed to connect to Hive/Impala via gohive v2.

type DSN

type DSN struct {
	Username          string
	Password          string
	Host              string
	Port              int
	Database          string
	Auth              string
	TransportMode     string
	Service           string
	SSLCertFile       string
	SSLKeyFile        string
	SSLInsecureSkip   bool
	HiveConfiguration map[string]string
}

DSN represents a parsed Data Source Name

func ParseDSN

func ParseDSN(dsn string) (*DSN, error)

ParseDSN parses a DSN string into a DSN struct

type Driver

type Driver struct{}

Driver is the interface that must be implemented by a database driver.

func (*Driver) Open

func (d *Driver) Open(name string) (driver.Conn, error)

Open returns a new connection to the database. The name is a string in a driver-specific format.

func (*Driver) OpenConnector

func (d *Driver) OpenConnector(name string) (driver.Connector, error)

OpenConnector implements driver.DriverContext

type HiveConnector

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

HiveConnector implements driver.Connector using gohive v2 under the hood.

func NewConnector

func NewConnector(cfg Config) *HiveConnector

NewConnector creates a new HiveConnector with the given config.

func (*HiveConnector) Connect

func (c *HiveConnector) Connect(ctx context.Context) (driver.Conn, error)

Connect establishes a new connection using gohive v2.

func (*HiveConnector) Driver

func (c *HiveConnector) Driver() driver.Driver

Driver returns a placeholder driver (unused by sql.OpenDB).

type HiveMetastoreClient

type HiveMetastoreClient struct {
	Client *hive_metastore.ThriftHiveMetastoreClient
	// contains filtered or unexported fields
}

func ConnectToMetastore

func ConnectToMetastore(host string, port int, auth string, configuration *MetastoreConnectConfiguration) (client *HiveMetastoreClient, err error)

Open connection to the metastore.

func (*HiveMetastoreClient) Close

func (c *HiveMetastoreClient) Close()

type MetastoreConnectConfiguration

type MetastoreConnectConfiguration struct {
	TransportMode string
	Username      string
	Password      string
}

func NewMetastoreConnectConfiguration

func NewMetastoreConnectConfiguration() *MetastoreConnectConfiguration

type TSaslTransport

type TSaslTransport struct {
	OpeningContext context.Context
	// contains filtered or unexported fields
}

TSaslTransport is a tranport thrift struct that uses SASL

func NewTSaslTransport

func NewTSaslTransport(trans thrift.TTransport, host string, mechanismName string, configuration map[string]string, maxLength uint32) (*TSaslTransport, error)

NewTSaslTransport return a TSaslTransport

func (*TSaslTransport) Close

func (p *TSaslTransport) Close() (err error)

Close close a SASL transport connection

func (*TSaslTransport) Flush

func (p *TSaslTransport) Flush(ctx context.Context) (err error)

Flush the bytes in the buffer

func (*TSaslTransport) IsOpen

func (p *TSaslTransport) IsOpen() bool

IsOpen opens a SASL connection

func (*TSaslTransport) Open

func (p *TSaslTransport) Open() (err error)

Open check if a SASL transport connection is opened

func (*TSaslTransport) Read

func (p *TSaslTransport) Read(buf []byte) (l int, err error)

func (*TSaslTransport) RemainingBytes

func (p *TSaslTransport) RemainingBytes() uint64

RemainingBytes return the size of the unwrapped bytes

func (*TSaslTransport) SetMaxLength

func (p *TSaslTransport) SetMaxLength(maxLength uint32)

SetMaxLength set the maxLength

func (*TSaslTransport) Write

func (p *TSaslTransport) Write(buf []byte) (int, error)