athena

README

go-athena

Note
Segment has paused maintenance on this project, but may return it to an active status in the future. Issues and pull requests from external contributors are not being considered, although internal contributions may appear from time to time. The project remains available under its open source license for anyone to use.

go-athena is a simple Golang database/sql driver for Amazon Athena.

import (
    "database/sql"
    _ "github.com/segmentio/go-athena"
)

func main() {
  db, _ := sql.Open("athena", "db=default&output_location=s3://results")
  rows, _ := db.Query("SELECT url, code from cloudfront")

  for rows.Next() {
    var url string
    var code int
    rows.Scan(&url, &code)
  }
}

It provides a higher-level, idiomatic wrapper over the AWS Go SDK, comparable to the Athena JDBC driver AWS provides for Java users.

For example,

• Instead of manually parsing types from strings, you can use database/sql.Rows.Scan()
• Instead of reaching for semaphores, you can use database/sql.DB.SetMaxOpenConns
• And, so on...

Caveats

database/sql exposes lots of methods that aren't supported in Athena. For example, Athena doesn't support transactions so Begin() is irrelevant. If a method must be supplied to satisfy a standard library interface but is unsupported, the driver will panic indicating so. If there are new offerings in Athena and/or helpful additions, feel free to PR.

Testing

Athena doesn't have a local version and revolves around S3 so our tests are integration tests against AWS itself. Thus, our tests require AWS credentials. The simplest way to provide them is via AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables, but you can use anything supported by the Default Credential Provider Chain.

The tests support a few environment variables:

• ATHENA_DATABASE can be used to override the default database "go_athena_tests"
• S3_BUCKET can be used to override the default S3 bucket of "go-athena-tests"

Documentation

Index

• Constants
• func EnableMockMode(mockAthenaClient athenaiface.AthenaAPI)
• func MockQuery(mocker Mocker, columnNames []string, columnTypes []string, ...)
• func Open(cfg Config) (*sql.DB, error)
• type AthenaAPI
• type AthenaDate
  • func (t AthenaDate) Equal(t2 AthenaDate) bool
  • func (t AthenaDate) MarshalJSON() ([]byte, error)
  • func (t AthenaDate) String() string
  • func (t AthenaDate) ToQueryValue() string
• type Config
• type Driver
  • func NewDriver(cfg *Config) *Driver
  • func (d *Driver) Open(connStr string) (driver.Conn, error)
• type Mocker
• type RowsIterator
  • func (r *RowsIterator[T]) GetData() (*T, error)
• type Stringer

Constants

const (
	// TimestampLayout is the Go time layout string for an Athena `timestamp`.
	TimestampLayout             = "2006-01-02 15:04:05.999"
	TimestampWithTimeZoneLayout = "2006-01-02 15:04:05.999 MST"
	DateLayout                  = "2006-01-02"
)

Functions

func EnableMockMode

func EnableMockMode(mockAthenaClient athenaiface.AthenaAPI)

EnableMockMode allows you to use a mock implementation of the Athena API.

func MockQuery

func MockQuery(mocker Mocker, columnNames []string, columnTypes []string, mockDataRows [][]string)

MockQuery mocks the AthenaAPI to return the given columns and data rows. e.g MockQuery(&mockAPI, map[string]string{"id": "string"}, [][]string{{"1"}})

func Open

func Open(cfg Config) (*sql.DB, error)

Open is a more robust version of `db.Open`, as it accepts a raw aws.Session. This is useful if you have a complex AWS session since the driver doesn't currently attempt to serialize all options into a string.

Types

type AthenaAPI

type AthenaAPI interface {
	athenaiface.AthenaAPI
}

AthenaAPI is an interface that represents the AthenaAPI. It's useful for mocking.

type AthenaDate

type AthenaDate time.Time

func (AthenaDate) Equal

func (t AthenaDate) Equal(t2 AthenaDate) bool

func (AthenaDate) MarshalJSON

func (t AthenaDate) MarshalJSON() ([]byte, error)

func (AthenaDate) String

func (t AthenaDate) String() string

func (AthenaDate) ToQueryValue

func (t AthenaDate) ToQueryValue() string

type Config

type Config struct {
	Session        *session.Session
	Database       string
	OutputLocation string

	PollFrequency      time.Duration
	PollRetryIncrement time.Duration
	MaxRetryDuration   time.Duration
	WorkGroup          *string
	DataCateLog        *string
}

Config is the input to Open().

type Driver

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

Driver is a sql.Driver. It's intended for db/sql.Open().

func NewDriver

func NewDriver(cfg *Config) *Driver

NewDriver allows you to register your own driver with `sql.Register`. It's useful for more complex use cases. Read more in PR #3. https://github.com/segmentio/go-athena/pull/3

Generally, sql.Open() or athena.Open() should suffice.

func (*Driver) Open

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

Open should be used via `db/sql.Open("athena", "<params>")`. The following parameters are supported in URI query format (k=v&k2=v2&...) example: "db=default&data_catalog=default&aws_access_key_id=default&aws_access_key_secrete=default&region=default&output_location=s3://results"

- `db` (required) refer to https://docs.aws.amazon.com/athena/latest/ug/understanding-tables-databases-and-the-data-catalog.html This is the Athena database name. In the UI, this defaults to "default", but the driver requires it regardless.

- `output_location` (required) This is the S3 location Athena will dump query results in the format "s3://bucket/and/so/forth". In the AWS UI, this defaults to "s3://aws-athena-query-results-<ACCOUNTID>-<REGION>", but the driver requires it.

- `poll_frequency` (optional) Athena's API requires polling to retrieve query results. This is the frequency at which the driver will poll for results. It should be a time/Duration.String(). A completely arbitrary default of "5s" was chosen.

- `work_group` (optional) Athena's API allows you to specify a workgroup for queries. This is the name of the workgroup you want to use. If not specified, the default workgroup is used.

- `data_catalog` (required) refer to https://docs.aws.amazon.com/athena/latest/ug/understanding-tables-databases-and-the-data-catalog.html Athena's API allows you to specify a data catalog for queries. This is the name of the data catalog you want to use. If not specified, the default data catalog is used.

- `region` (required) Override AWS region. Useful if it is not set with environment variable.

- `aws_access_key_id` (required) AWS access key id. Useful if it is not set with environment variable.

- `aws_access_key_secret` (required) AWS access key secret. Useful if it is not set with environment variable. Credentials must be accessible via the SDK's Default Credential Provider Chain. For more advanced AWS credentials/session/config management, please supply a custom AWS session directly via `athena.Open()`.

type Mocker

type Mocker interface {
	On(methodName string, arguments ...interface{}) *mock.Call
}

type RowsIterator

type RowsIterator[T any] struct {
	Rows *sqlx.Rows
	Cnt  int
}

RowsIterator is a struct that helps to iterate over the rows of the query result

func (*RowsIterator[T]) GetData

func (r *RowsIterator[T]) GetData() (*T, error)

GetData returns the next row of the query result

type Stringer

type Stringer interface {
	String() string
}