rin

README

Rin

Rin is a Redshift data Importer by SQS messaging.

Architecture

1. (Someone) creates a S3 object.
2. S3 event notifications will send to a message to SQS.
3. Rin will fetch messages from SQS, and publish a "COPY" query to Redshift.

Configuration

Configuring Amazon S3 Event Notifications.

1. Create SQS queue.
2. Attach SQS access policy to the queue. Example Walkthrough 1:
3. Enable Event Notifications on a S3 bucket.
4. Run rin process with configuration for using the SQS and S3.

config.yaml

queue_name: my_queue_name    # SQS queue name

credentials:
  aws_access_key_id: AAA
  aws_secret_access_key: SSS
  aws_region: ap-northeast-1

redshift:
  host: localhost
  port: 5439
  dbname: test
  user: test_user
  password: test_pass
  schema: public

s3:
  bucket: test.bucket.test
  region: ap-northeast-1

sql_option: "JSON 'auto' GZIP"       # COPY SQL option

# define import target mappings
targets:
  - redshift:
      table: foo
    s3:
      key_prefix: test/foo

  - redshift:
      schema: xxx
      table: bar
    s3:
      key_prefix: test/bar

  - redshift:
      schema: $1      # expand by key_regexp captured value.
      table: $2
    s3:
      key_regexp: test/schema-([a-z]+)/table-([a-z]+)/

  - redshift:
      host: redshift.example.com       # override default section in this target
      port: 5439
      dbname: example
      user: example_user
      password: example_pass
      schema: public
      table: example
    s3:
      bucket: redshift.example.com
      region: ap-northeast-1
      key_prefix: logs/example/
    sql_option: "CSV DELIMITER ',' ESCAPE"

Credentials

Rin requires credentials for SQS and Redshift.

1. credentials.aws_access_key_id and credentials.aws_secret_access_key

• use for SQS and Redshift.

1. credentials.aws_iam_role

• use for Redshift only.
• for SQS, Rin will try to get a instance credentials.

Run

daemon mode

Rin waits new SQS messages and processing it continually.

$ rin -config config.yaml [-debug]

batch mode

Rin process new SQS messages and exit.

$ rin -config config.yaml -batch [-debug]

Documentation

Index

• Constants
• Variables
• func ConnectToRedshift(target *Target) (*sql.DB, error)
• func Import(event Event) (int, error)
• func ImportRedshift(target *Target, record *EventRecord, cap *[]string) error
• func Run(configFile string, batchMode bool) error
• type AuthExpiration
  • func (e AuthExpiration) Error() string
  • func (e AuthExpiration) Signal()
  • func (e AuthExpiration) String() string
• type Config
  • func LoadConfig(path string) (*Config, error)
• type Credentials
  • func (c Credentials) RedshiftCredential() string
• type Event
  • func ParseEvent(b []byte) (Event, error)
  • func (e Event) String() string
• type EventRecord
  • func (r EventRecord) String() string
• type NoMessageError
  • func (e NoMessageError) Error() string
• type Redshift
  • func (r Redshift) DSN() string
  • func (r Redshift) String() string
  • func (r Redshift) VisibleDSN() string
• type S3
  • func (s3 S3) String() string
• type S3Bucket
• type S3Event
• type S3Object
• type SQLParam
• type Target
  • func (t *Target) BuildCopySQL(key string, cred Credentials, cap *[]string) (string, error)
  • func (t *Target) Match(bucket, key string) (bool, *[]string)
  • func (t *Target) MatchEventRecord(r *EventRecord) (bool, *[]string)
  • func (t *Target) String() string

Constants

const (
	S3URITemplate = "s3://%s/%s"
	SQLTemplate   = "/* Rin */ COPY %s FROM %s CREDENTIALS '%s' REGION '%s' %s"
)

Variables

var (
	DBPool      = make(map[string]*sql.DB, 0)
	DBPoolMutex sync.Mutex
)

var Debug bool

var MaxDeleteRetry = 8

var Runnable bool

var SQS *sqs.SQS

var TrapSignals = []os.Signal{
	syscall.SIGHUP,
	syscall.SIGINT,
	syscall.SIGTERM,
	syscall.SIGQUIT,
}

Functions

func ConnectToRedshift

func ConnectToRedshift(target *Target) (*sql.DB, error)

func Import

func Import(event Event) (int, error)

func ImportRedshift

func ImportRedshift(target *Target, record *EventRecord, cap *[]string) error

func Run

func Run(configFile string, batchMode bool) error

Types

type AuthExpiration

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

func (AuthExpiration) Error

func (e AuthExpiration) Error() string

func (AuthExpiration) Signal

func (e AuthExpiration) Signal()

func (AuthExpiration) String

func (e AuthExpiration) String() string

type Config

type Config struct {
	QueueName   string      `yaml:"queue_name"`
	Targets     []*Target   `yaml:"targets"`
	Credentials Credentials `yaml:"credentials"`
	Redshift    *Redshift   `yaml:"redshift"`
	S3          *S3         `yaml:"s3"`
	SQLOption   string      `yaml:"sql_option"`
}

func LoadConfig

func LoadConfig(path string) (*Config, error)

type Credentials

type Credentials struct {
	AWS_ACCESS_KEY_ID     string `yaml:"aws_access_key_id"`
	AWS_SECRET_ACCESS_KEY string `yaml:"aws_secret_access_key"`
	AWS_REGION            string `yaml:"aws_region"`
	AWS_IAM_ROLE          string `yaml:"aws_iam_role"`
}

func (Credentials) RedshiftCredential

func (c Credentials) RedshiftCredential() string

type Event

type Event struct {
	Records []*EventRecord `json:"Records"`
}

func ParseEvent

func ParseEvent(b []byte) (Event, error)

func (Event) String

func (e Event) String() string

type EventRecord

type EventRecord struct {
	EventVersion string  `json:"eventVersion"`
	EventName    string  `json:"eventName"`
	EventSource  string  `json:"eventSource"`
	EventTime    string  `json:"eventTime"`
	AWSRegion    string  `json:"awsRegion"`
	S3           S3Event `json:"s3"`
}

func (EventRecord) String

func (r EventRecord) String() string

type NoMessageError

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

func (NoMessageError) Error

func (e NoMessageError) Error() string

type Redshift

type Redshift struct {
	Host     string `yaml:"host"`
	Port     int    `yaml:"port"`
	DBName   string `yaml:"dbname"`
	User     string `yaml:"user"`
	Password string `yaml:"password"`
	Schema   string `yaml:"schema"`
	Table    string `yaml:"table"`
}

func (Redshift) DSN

func (r Redshift) DSN() string

func (Redshift) String

func (r Redshift) String() string

func (Redshift) VisibleDSN

func (r Redshift) VisibleDSN() string

type S3

type S3 struct {
	Region    string `yaml:"region"`
	Bucket    string `yaml:"bucket"`
	KeyPrefix string `yaml:"key_prefix"`
	KeyRegexp string `yaml:"key_regexp"`
}

func (S3) String

func (s3 S3) String() string

type S3Bucket

type S3Bucket struct {
	Name string `json:"name"`
	ARN  string `json:"arn"`
}

type S3Event

type S3Event struct {
	S3SchemaVersion string   `json:"s3SchemaVersion"`
	ConfigurationID string   `json:"configurationId"`
	Bucket          S3Bucket `json:"bucket"`
	Object          S3Object `json:"object"`
}

type S3Object

type S3Object struct {
	Key  string `json:"key"`
	Size int64  `json:"size"`
	ETag string `json:"eTag"`
}

type SQLParam

type SQLParam struct {
	Table  string
	Option string
}

type Target

type Target struct {
	Redshift  *Redshift `yaml:"redshift"`
	S3        *S3       `yaml:"s3"`
	SQLOption string    `yaml:"sql_option"`
	// contains filtered or unexported fields
}

func (*Target) BuildCopySQL

func (t *Target) BuildCopySQL(key string, cred Credentials, cap *[]string) (string, error)

func (*Target) Match

func (t *Target) Match(bucket, key string) (bool, *[]string)

func (*Target) MatchEventRecord

func (t *Target) MatchEventRecord(r *EventRecord) (bool, *[]string)

func (*Target) String

func (t *Target) String() string