firehose

README

firehose

Firehose has the following functionality:

1. CreateDeliveryStream
2. PutRecord
3. PutRecordBatch
4. DescribeDeliveryStream
5. DeleteDeliveryStream

Use cases

1- insert single record to existing delivery stream: - first you will need to create new firehose client - use PutRecord to insert a single record to specific stream, which take 2 params channel and record

2- insert bulk records to new delivery stream: - first you will need to create new firehose client - then create new delivery stream using CreateDeliveryStream, which take 1 param channelName - use PutRecordBatch to insert multiple records to specific stream, which take 2 params channel and bulkrecords PutRecordBatch will handle size limitations internally, by dividing batch to smaller batches with proper size.

3- check existence of delivery stream: - use DescribeDeliveryStream if status is ACTIVE therefore ready for data to be sent to it if status is CREATING_FAILED, so you need to delete deliver stream and create it again

Documentation

Index

• type ClientProvider
  • func NewClientProvider() (*ClientProvider, error)
  • func (c *ClientProvider) CreateDeliveryStream(channel string) error
  • func (c *ClientProvider) DeleteDeliveryStream(channel string, force bool) error
  • func (c *ClientProvider) DescribeDeliveryStream(channel string) (*DescribeOutput, error)
  • func (c *ClientProvider) PutRecord(channel string, record interface{}) (*PutResponse, error)
  • func (c *ClientProvider) PutRecordBatch(channel string, records []interface{}) ([]*PutResponse, error)
• type Config
• type DescribeOutput
• type PutResponse

Types

type ClientProvider

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

ClientProvider for kinesis firehose

func NewClientProvider

func NewClientProvider() (*ClientProvider, error)

NewClientProvider initiate new client provider

func (*ClientProvider) CreateDeliveryStream

func (c *ClientProvider) CreateDeliveryStream(channel string) error

CreateDeliveryStream creating firehose delivery stream channel You must provide channel name as required parameter If channel created successfully it will return nil else it will return error

func (*ClientProvider) DeleteDeliveryStream

func (c *ClientProvider) DeleteDeliveryStream(channel string, force bool) error

DeleteDeliveryStream ... Deletes a delivery stream and its data. To check the state of a delivery stream, use DescribeDeliveryStream. You can delete a delivery stream only if it is in one of the following states: ACTIVE, DELETING, CREATING_FAILED, or DELETING_FAILED. You can't delete a delivery stream that is in the CREATING state. While the deletion request is in process, the delivery stream is in the DELETING state. While the delivery stream is in the DELETING state, the service might continue to accept records, but it doesn't make any guarantees with respect to delivering the data. Therefore, as a best practice, first stop any applications that are sending records before you delete a delivery stream.

func (*ClientProvider) DescribeDeliveryStream

func (c *ClientProvider) DescribeDeliveryStream(channel string) (*DescribeOutput, error)

DescribeDeliveryStream Describes the specified delivery stream and its status. For example, after your delivery stream is created, call DescribeDeliveryStream to see whether the delivery stream is ACTIVE and therefore ready for data to be sent to it. If the status of a delivery stream is CREATING_FAILED, this status doesn't change, and you can't invoke CreateDeliveryStream again on it. However, you can invoke the DeleteDeliveryStream operation to delete it. If the status is DELETING_FAILED, you can force deletion by invoking DeleteDeliveryStream again but with DeleteDeliveryStreamInput$AllowForceDelete set to true.

func (*ClientProvider) PutRecord

func (c *ClientProvider) PutRecord(channel string, record interface{}) (*PutResponse, error)

PutRecord is operation for Amazon Kinesis Firehose. Writes a single data record into an Amazon Kinesis Data Firehose delivery stream.

By default, each delivery stream can take in up to 2,000 transactions per second, 5,000 records per second, or 5 MB per second.

You must specify the name of the delivery stream and the data record when using PutRecord. The data record consists of a data blob that can be up to 1,000 KB in size, and any kind of data. You must specify the name of the delivery stream and the data record when using PutRecord. The data record consists of a data blob that can be up to 1,000 KB in size, and any kind of data.

Kinesis Data Firehose buffers records before delivering them to the destination. To disambiguate the data blobs at the destination, a common solution is to use delimiters in the data, such as a newline (\n) or some other character unique within the data. This allows the consumer application to parse individual data items when reading the data from the destination.

The PutRecord operation returns a RecordId, which is a unique string assigned to each record.

func (*ClientProvider) PutRecordBatch

func (c *ClientProvider) PutRecordBatch(channel string, records []interface{}) ([]*PutResponse, error)

PutRecordBatch is operation for Amazon Kinesis Firehose Writes multiple data records into a delivery stream in a single call, which can achieve higher throughput per producer than when writing single records.

Each PutRecordBatch request supports up to 500 records. Each record in the request can be as large as 1,000 KB (before 64-bit encoding), up to a limit of 4 MB for the entire request.

You must specify the name of the delivery stream and the data record when using PutRecord. The data record consists of a data blob that can be up to 1,000 KB in size.

The PutRecordBatch response includes a map of failed records. Even if the PutRecordBatch call succeeds

Data records sent to Kinesis Data Firehose are stored for 24 hours from the time they are added to a delivery stream as it attempts to send the records to the destination. If the destination is unreachable for more than 24 hours, the data is no longer available.

Don't concatenate two or more base64 strings to form the data fields of your records. Instead, concatenate the raw data, then perform base64 encoding.

type Config

type Config struct {
	Endpoint string
	Region   string
}

Config aws configuration

type DescribeOutput

type DescribeOutput struct {
	StreamStatus string
}

DescribeOutput ...

type PutResponse

type PutResponse struct {
	RecordID string
	Error    error
}

PutResponse per record