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

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

        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

          func GetIngestionFailureStatus(err error) (FailureStatusCode, error)

            GetIngestionFailureStatus extracts the ingestion failure code from an ingestion error

            func (FailureStatusCode) IsRetryable

            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

                              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
                                          Output:
                                          
                                          

                                          func (*Ingestion) FromReader

                                          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

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

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

                                                func (*Result) Wait

                                                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

                                                  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

                                                    func GetIngestionStatus(err error) (StatusCode, error)

                                                      GetIngestionStatus extracts the ingestion status code from an ingestion error

                                                      func (StatusCode) IsFinal

                                                      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

                                                        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
                                                                )

                                                                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.