ingest

package
v0.3.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 3, 2021 License: MIT Imports: 20 Imported by: 9

Documentation

Overview

Package ingest provides data ingestion from various external sources into Kusto.

For more information on Kusto Data Ingestion, please see: https://docs.microsoft.com/en-us/azure/kusto/management/data-ingestion/

Create a client

Creating a client simply requires a *kusto.Client, the name of the database and the name of the table to be ingested into. 

in, err := ingest.New(kustoClient, "database", "table")
if err != nil {
	panic("add error handling")
}

Ingestion from a local file

Ingesting a local file requires simply passing the path to the file to be ingested: 

if _, err := in.FromFile(ctx, "/path/to/a/local/file"); err != nil {
	panic("add error handling")
}

FromFile() will accept Unix path names on Unix platforms and Windows path names on Windows platforms. The file will not be deleted after upload (there is an option that will allow that though).

Ingestion from an Azure Blob Storage file

This package will also accept ingestion from an Azure Blob Storage file: 

if _, err := in.FromFile(ctx, "https://myaccount.blob.core.windows.net/$root/myblob"); err != nil {
	panic("add error handling")
}

This will ingest a file from Azure Blob Storage. We only support https:// paths and your domain name may differ than what is here.

Ingestion from an io.Reader

Sometimes you want to ingest a stream of data that you have in memory without writing to disk. You can do this simply by chunking the data via an io.Reader. 

r, w := io.Pipe()

enc := json.NewEncoder(w)
go func() {
	defer w.Close()
	for _, data := range dataSet {
		if err := enc.Encode(data); err != nil {
			panic("add error handling")
		}
	}
}()

if _, err := in.FromReader(ctx, r); err != nil {
	panic("add error handling")
}

It is important to remember that FromReader() will terminate when it receives an io.EOF from the io.Reader. Use io.Readers that won't return io.EOF until the io.Writer is closed (such as io.Pipe).

Ingestion from a Stream

Instestion from a stream commits blocks of fully formed data encodes (JSON, AVRO, ...) into Kusto: 

if err := in.Stream(ctx , jsonEncodedData, ingest.JSON, "mappingName"); err != nil {
	panic("add error handling")
}

Ingestion with Status Reporting

You can use Kusto Go SDK to get table-based status reporting of ingestion operations. Ingestion commands run using FromFile() and FromReader() return an error and a channel that can be waited upon for a final status. If the error is not nil, the operation has failed locally. If the error is nil and Table Status Reporting option was used, the SDK user can wait on the channel for a success (nil) or failure (Error) status.

Note! This feature is not suitable for users running ingestion at high rates, and may slow down the ingestion operation. 

status, err := ingestor.FromFile(ctx, "/path/to/file", ingest.ReportResultToTable())
if err != nil {
	// The ingestion command failed to be sent, Do something
}

err = <-status.Wait(ctx)
if err != nil {
	// the operation complete with an error
	if ingest.IsRetryable(err) {
		// Handle retries
	} else {
		// inspect the failure
		// statusCode, _ := ingest.GetIngestionStatus(err)
		// failureStatus, _ := ingest.GetIngestionFailureStatus(err)
	}
}

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// ErrTooLarge indicates that the data being passed to a StreamBlock is larger than the maximum StreamBlock size of 4MiB.
	ErrTooLarge = errors.ES(errors.OpIngestStream, errors.KClientArgs, "cannot add data larger than 4MiB")
)

Functions

func IsRetryable added in v0.3.0 

func IsRetryable(err error) bool

IsRetryable indicates whether there's any merit in retying ingestion

Types

type DataFormat 

type DataFormat = properties.DataFormat

DataFormat indicates what type of encoding format was used for source data. Not all options can be used in every method. 

const (
	// DFUnknown indicates the EncodingType is not set.
	DFUnknown DataFormat = properties.DFUnknown
	// AVRO indicates the source is encoded in Apache Avro format.
	AVRO DataFormat = properties.AVRO
	// ApacheAVRO indicates the source is encoded in Apache avro2json format.
	ApacheAVRO DataFormat = properties.ApacheAVRO
	// CSV indicates the source is encoded in comma seperated values.
	CSV DataFormat = properties.CSV
	// JSON indicates the source is encoded as one or more lines, each containing a record in Javscript Object Notation.
	JSON DataFormat = properties.JSON
	// MultiJSON indicates the source is encoded in JSON-Array of individual records in Javscript Object Notation.
	MultiJSON DataFormat = properties.MultiJSON
	// ORC indicates the source is encoded in Apache Optimized Row Columnar format.
	ORC DataFormat = properties.ORC
	// Parquet indicates the source is encoded in Apache Parquet format.
	Parquet DataFormat = properties.Parquet
	// PSV is pipe "|" separated values.
	PSV DataFormat = properties.PSV
	// Raw is a text file that has only a single string value.
	Raw DataFormat = properties.Raw
	// SCSV is a file containing semicolon ";" separated values.
	SCSV DataFormat = properties.SCSV
	// SOHSV is a file containing SOH-separated values(ASCII codepont 1).
	SOHSV DataFormat = properties.SOHSV
	// SStream indicats the source is encoded as a Microsoft Cosmos Structured Streams format
	SStream DataFormat = properties.SStream
	// TSV is a file containing tab seperated values ("\t").
	TSV DataFormat = properties.TSV
	// TSVE is a file containing escaped-tab seperated values ("\t").
	TSVE DataFormat = properties.TSVE
	// TXT is a text file with lines deliminated by "\n".
	TXT DataFormat = properties.TXT
	// WCLogFile indicates the source is encoded using W3C Extended Log File format
	W3CLogFile DataFormat = properties.W3CLogFile
)

note: any change here needs to be kept up to date with the properties version. I'm not a fan of having two copies, but I don't think it is worth moving to its own package to allow properties and ingest to both import without a cycle.

type FailureStatusCode added in v0.3.0 

type FailureStatusCode string

FailureStatusCode indicates the status of failuted ingestion attempts 

const (
	// Unknown represents an undefined or unset failure state
	Unknown FailureStatusCode = "Unknown"
	// Permanent represnets failure state that will benefit from a retry attempt
	Permanent FailureStatusCode = "Permanent"
	// Transient represnet a retryable failure state
	Transient FailureStatusCode = "Transient"
	// Exhausted represents a retryable failure that has exhusted all retry attempts
	Exhausted FailureStatusCode = "Exhausted"
)

func GetIngestionFailureStatus added in v0.3.0 

func GetIngestionFailureStatus(err error) (FailureStatusCode, error)

GetIngestionFailureStatus extracts the ingestion failure code from an ingestion error

func (FailureStatusCode) IsRetryable added in v0.3.0 

func (i FailureStatusCode) IsRetryable() bool

IsRetryable indicates whether there's any merit in retying ingestion

type FileOption 

type FileOption interface {
	// contains filtered or unexported methods
}

FileOption is an optional argument to FromFile().

func DeleteSource 

func DeleteSource() FileOption

DeleteSource deletes the source file from when it has been uploaded to Kusto.

func FileFormat 

func FileFormat(et DataFormat) FileOption

FileFormat can be used to indicate what type of encoding is supported for the file. This is only needed if the file extension is not present. A file like: "input.json.gz" or "input.json" does not need this option, while "input" would.

func FlushImmediately 

func FlushImmediately() FileOption

FlushImmediately tells Kusto to flush on write.

func IfNotExists 

func IfNotExists(ingestByTag string) FileOption

IfNotExists provides a string value that, if specified, prevents ingestion from succeeding if the table already has data tagged with an ingest-by: tag with the same value. This ensures idempotent data ingestion. For more information see: https://docs.microsoft.com/en-us/azure/kusto/management/extents-overview#ingest-by-extent-tags

func IgnoreSizeLimit 

func IgnoreSizeLimit() FileOption

IgnoreSizeLimit ignores the size limit for data ingestion.

func IngestionMapping 

func IngestionMapping(mapping interface{}, mappingKind DataFormat) FileOption

IngestionMapping provides runtime mapping of the data being imported to the fields in the table. "ref" will be JSON encoded, so it can be any type that can be JSON marshalled. If you pass a string or []byte, it will be interpreted as already being JSON encoded. mappingKind can only be: CSV, JSON, AVRO, Parquet or ORC.

func IngestionMappingRef 

func IngestionMappingRef(refName string, mappingKind DataFormat) FileOption

IngestionMappingRef provides the name of a pre-created mapping for the data being imported to the fields in the table. mappingKind can only be: CSV, JSON, AVRO, Parquet or ORC. For more details, see: https://docs.microsoft.com/en-us/azure/kusto/management/create-ingestion-mapping-command

func ReportResultToTable added in v0.3.0 

func ReportResultToTable() FileOption

ReportResultToTable option requests that the ingestion status will be tracked in an Azure table. Note using Table status reporting is not recommended for high capacity ingestions, as it could slow down the ingestion. In such cases, it's recommended to enable it temporarily for debugging failed ingestions.

func Tags 

func Tags(tags []string) FileOption

Tags are tags to be associated with the ingested ata.

func ValidationPolicy 

func ValidationPolicy(policy ValPolicy) FileOption

ValidationPolicy uses a ValPolicy to set our ingestion data validation policy. If not set, no validation policy is used. For more information, see: https://docs.microsoft.com/en-us/azure/kusto/management/data-ingestion/

type Ingestion 

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

Ingestion provides data ingestion from external sources into Kusto.

func New 

func New(client *kusto.Client, db, table string) (*Ingestion, error)

New is the constructor for Ingestion.

func (*Ingestion) FromFile 

func (i *Ingestion) FromFile(ctx context.Context, fPath string, options ...FileOption) (*Result, error)

FromFile allows uploading a data file for Kusto from either a local path or a blobstore URI path. This method is thread-safe.

Example 
package main

import (
	"context"
	"time"

	"github.com/Azure/azure-kusto-go/kusto"
	"github.com/Azure/azure-kusto-go/kusto/ingest"
	"github.com/Azure/go-autorest/autorest/azure/auth"
)

func main() {
	var err error

	authConfig := auth.NewClientCredentialsConfig("clientID", "clientSecret", "tenantID")
	/*
		Alteratively, you could so something like:
		authorizer, err := auth.NewMSIConfig().Authorizer()
		or
		authorizer, err := auth.NewAuthorizerFromEnvironment()
		or
		auth.New...()

		then
		kusto.Authorization{Authorizer: authorizer}
	*/

	client, err := kusto.New("endpoint", kusto.Authorization{Config: authConfig})
	if err != nil {
		// Do something
	}

	ingestor, err := ingest.New(client, "database", "table")
	if err != nil {
		// Do something
	}

	// Setup a maximum time for completion to be 10 minutes.
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
	defer cancel()

	// Upload our file WITHOUT status reporting.
	// When completed, delete the file on local storage we are uploading.
	_, err = ingestor.FromFile(ctx, "/path/to/file", ingest.DeleteSource())
	if err != nil {
		// The ingestion command failed to be sent, Do something
	}

	// Upload our file WITH status reporting.
	// When completed, delete the file on local storage we are uploading.
	status, err := ingestor.FromFile(ctx, "/path/to/file", ingest.DeleteSource(), ingest.ReportResultToTable())
	if err != nil {
		// The ingestion command failed to be sent, Do something
	}

	err = <-status.Wait(ctx)
	if err != nil {
		// the operation complete with an error
		if ingest.IsRetryable(err) {
			// Handle reties
		} else {
			// inspect the failure
			// statusCode, _ := ingest.GetIngestionStatus(err)
			// failureStatus, _ := ingest.GetIngestionFailureStatus(err)
		}
	}
}

Output:

func (*Ingestion) FromReader added in v0.2.0 

func (i *Ingestion) FromReader(ctx context.Context, reader io.Reader, options ...FileOption) (*Result, error)

FromReader allows uploading a data file for Kusto from an io.Reader. The content is uploaded to Blobstore and ingested after all data in the reader is processed. Content should not use compression as the content will be compressed with gzip. This method is thread-safe.

func (*Ingestion) Stream 

func (i *Ingestion) Stream(ctx context.Context, payload []byte, format DataFormat, mappingName string) error

Stream takes a payload that is encoded in format with a server stored mappingName, compresses it and uploads it to Kusto. payload must be a fully formed entry of format and < 4MiB or this will fail. We currently support CSV, TSV, SCSV, SOHSV, PSV, JSON and AVRO. If using JSON or AVRO, you must provide a mappingName that references the name of the pre-created ingestion mapping defined on the table. Otherwise mappingName can be an empty string. More information can be found here: https://docs.microsoft.com/en-us/azure/kusto/management/create-ingestion-mapping-command The context object can be used with a timeout or cancel to limit the request time.

type Result added in v0.3.0 

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

Result provides a way for users track the state of ingestion jobs.

func (*Result) Wait added in v0.3.0 

func (r *Result) Wait(ctx context.Context) chan error

Wait returns a channel that can be checked for ingestion results. In order to check actual status please use the IngestionStatus option when ingesting data.

type StatusCode added in v0.3.0 

type StatusCode string

StatusCode is the ingestion status 

const (
	// Pending status represents a temporary status.
	// Might change during the course of ingestion based on the
	// outcome of the data ingestion operation into Kusto.
	Pending StatusCode = "Pending"
	// Succeeded status represents a permanent status.
	// The data has been successfully ingested to Kusto.
	Succeeded StatusCode = "Succeeded"
	// Failed Status represents a permanent status.
	// The data has not been ingested to Kusto.
	Failed StatusCode = "Failed"
	// Queued status represents a permanent status.
	// The data has been queued for ingestion &  status tracking was not requested.
	// (This does not indicate that the ingestion was successful.)
	Queued StatusCode = "Queued"
	// Skipped status represents a permanent status.
	// No data was supplied for ingestion. The ingest operation was skipped.
	Skipped StatusCode = "Skipped"
	// PartiallySucceeded status represents a permanent status.
	// Part of the data was successfully ingested to Kusto, while other parts failed.
	PartiallySucceeded StatusCode = "PartiallySucceeded"

	// StatusRetrievalFailed means the client ran into truble reading the status from the service
	StatusRetrievalFailed StatusCode = "StatusRetrievalFailed"
	// StatusRetrievalCanceled means the user canceld the status check
	StatusRetrievalCanceled StatusCode = "StatusRetrievalCanceled"
)

func GetIngestionStatus added in v0.3.0 

func GetIngestionStatus(err error) (StatusCode, error)

GetIngestionStatus extracts the ingestion status code from an ingestion error

func (StatusCode) IsFinal added in v0.3.0 

func (i StatusCode) IsFinal() bool

IsFinal returns true if the ingestion status is a final status, or false if the status is temporary

func (StatusCode) IsSuccess added in v0.3.0 

func (i StatusCode) IsSuccess() bool

IsSuccess returns true if the status code is a final successfull status code

type ValPolicy 

type ValPolicy struct {
	// Options provides an option that will flag data that does not validate.
	Options ValidationOption `json:"ValidationOptions"`
	// Implications sets what to do when a policy option is violated.
	Implications ValidationImplication `json:"ValidationImplications"`
}

ValPolicy sets a policy for validating data as it is sent for ingestion. For more information, see: https://docs.microsoft.com/en-us/azure/kusto/management/data-ingestion/

type ValidationImplication 

type ValidationImplication int8

ValidationImplication is a setting used to indicate what to do when a Validation Policy is violated. These are defined as constants within this package. 

const (
	// FailIngestion indicates that any violation of the ValidationPolicy will cause the entire ingestion to fail.
	FailIngestion ValidationImplication = 0
	// IgnoreFailures indicates that failure of the ValidationPolicy will be ignored.
	IgnoreFailures ValidationImplication = 1
)

type ValidationOption 

type ValidationOption int8

ValidationOption is an an option for validating the ingestion input data. These are defined as constants within this package. 

const (
	// VOUnknown indicates that a ValidationOption was not set.
	VOUnknown ValidationOption = 0
	// SameNumberOfFields indicates that all records ingested must have the same number of fields.
	SameNumberOfFields ValidationOption = 1
	// IgnoreNonDoubleQuotedFields indicates that fields that do not have double quotes should be ignored.
	IgnoreNonDoubleQuotedFields ValidationOption = 2
)

Source Files

View all Source files

Directories

Path Synopsis
internal
conn
Package conn holds a streaming ingest connetion.
 Package conn holds a streaming ingest connetion.
filesystem
Package filesystem provides a client with the ability to import data into Kusto via a variety of fileystems such as local storage or blobstore.
 Package filesystem provides a client with the ability to import data into Kusto via a variety of fileystems such as local storage or blobstore.
gzip
Package gzip provides a streaming object for taking in io.ReadCloser that is being written to and providing an io.ReadCloser that outputs the original content gzip compressed.
 Package gzip provides a streaming object for taking in io.ReadCloser that is being written to and providing an io.ReadCloser that outputs the original content gzip compressed.
properties
Package properties provides Kusto REST properties that will need to be serialized and sent to Kusto based upon the type of ingestion we are doing.
 Package properties provides Kusto REST properties that will need to be serialized and sent to Kusto based upon the type of ingestion we are doing.
resources
Package resources contains objects that are used to gather information about Kusto resources that are used during various ingestion methods.
 Package resources contains objects that are used to gather information about Kusto resources that are used during various ingestion methods.
status

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.