udf

package

v0.96.0 Latest Latest Go to latest Published: Oct 21, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/viant/endly

Links

Open Source Insights

README ¶

#User Defined Function Service

This service enables custom UDF registration with pre defined provider set.

Usage
Predefined UDF Providers
Service Actions

Usage

Register UDF

endly -r=reqister

@register.yaml

pipeline:
    register-udf:
      action: udf:register
      udfs:
        - id: AddressBookToProto
          provider: ProtoWriter
          params:
            - /Project/proto/address_book.proto
            - AddressBook
        - id: ProtoToAddressBook
          provider: ProtoReader
          params:
            - /Project/proto/address_book.proto
            - AddressBook

Avro Reader UDF data validation:

endly -r=test

@test.yaml

pipeline:
  loadData:
    action: storage:download
    udf: AvroReader
    source:
      URL: gs://mye2ebucket/data/output/1/app_data00000.avro
    credentials: $gcpSecrets
    destKey: matchedData
    post:
      myData: $Transformed

  infoMyData:
    action: print
    message: $AsJSON(${myData})

  infoMatched:
    action: print
    message: $AsJSON(${matchedData})

  infoAll:
    action: print
    message: $AsJSON(${loadData})
    
  assert:
    action: validator:assert
    actual: $AsData(${matchedData})
    expect: $Cat(${parent.path}/expect/mydata.json)

Predefined UDF Providers

Provider	Arguments
ProtoReader	schemaFile, messageType, importPath
ProtoReader	schemaFile, messageType, importPath
AvroWriter	avroSchema/URL, compression
CsvReader	headerFields, delimiter

Service actions

UDF Service

Service Id	Action	Description	Request	Response
udf	register	register custom UDF with udf provider	RegisterRequest	RegisterResponse

Documentation ¶

Index ¶

Constants
func AsProtobufMessage(source interface{}, state data.Map, target proto.Message) (interface{}, error)
func DateOfBirth(source interface{}, state data.Map) (interface{}, error)
func FromProtobufMessage(source interface{}, state data.Map, sourceMessage proto.Message) (interface{}, error)
func GZipContentCorrupter(source interface{}, state data.Map) (interface{}, error)
func GZipper(source interface{}, state data.Map) (interface{}, error)
func Hostname(source interface{}, state data.Map) (interface{}, error)
func LoadData(source interface{}, state data.Map) (interface{}, error)
func New() endly.Service
func NewAvroReader(source interface{}, state data.Map) (interface{}, error)
func NewAvroWriter(args ...interface{}) (func(source interface{}, state data.Map) (interface{}, error), error)
func NewCsvReader(args ...interface{}) (func(source interface{}, state data.Map) (interface{}, error), error)
func NewProtoReader(args ...interface{}) (func(source interface{}, state data.Map) (interface{}, error), error)
func NewProtoWriter(args ...interface{}) (func(source interface{}, state data.Map) (interface{}, error), error)
func RegisterProviders(providers []*endly.UdfProvider) error
func TransformWithUDF(context *endly.Context, udfName, source string, payload interface{}) (interface{}, error)
func URLJoin(source interface{}, state data.Map) (interface{}, error)
func URLPath(source interface{}, state data.Map) (interface{}, error)
type ProtoCodec
- func NewProtoCodec(schemaFile, importPath string, msgType string, lowercaseKey bool) (*ProtoCodec, error)
- func (c *ProtoCodec) AsBinary(msgType string, msg interface{}) ([]byte, error)
- func (c *ProtoCodec) AsMessage(msgType string, data []byte) (interface{}, error)
type RegisterRequest
- func NewRegisterRequestFromURL(URL string) (*RegisterRequest, error)
type RegisterResponse

Constants ¶

View Source

const (
	//ServiceID represents UDF service id.
	ServiceID = "udf"
)

Variables ¶

This section is empty.

Functions ¶

func AsProtobufMessage ¶

func AsProtobufMessage(source interface{}, state data.Map, target proto.Message) (interface{}, error)

AsProtobufMessage generic method for converting a map, or json string into a proto message

func DateOfBirth ¶

func DateOfBirth(source interface{}, state data.Map) (interface{}, error)

DateOfBirth returns formatted date of birth, it take desired age, [month], [day], [timeformat]

func FromProtobufMessage ¶

func FromProtobufMessage(source interface{}, state data.Map, sourceMessage proto.Message) (interface{}, error)

FromProtobufMessage generic method for converting a proto message into a map

func GZipContentCorrupter ¶

func GZipContentCorrupter(source interface{}, state data.Map) (interface{}, error)

GZipContentCorrupter corrupt zip content modifier

func GZipper ¶

func GZipper(source interface{}, state data.Map) (interface{}, error)

GZipper copy modifier, mofidies source using zip udf

func Hostname ¶

func Hostname(source interface{}, state data.Map) (interface{}, error)

Hostname return host from URL

func LoadData ¶

func LoadData(source interface{}, state data.Map) (interface{}, error)

func New ¶

func New() endly.Service

New creates a new udf service.

func NewAvroReader ¶

func NewAvroReader(source interface{}, state data.Map) (interface{}, error)

NewAvroReader creates a new avro reader UDFs

func NewAvroWriter ¶

func NewAvroWriter(args ...interface{}) (func(source interface{}, state data.Map) (interface{}, error), error)

NewAvroWriter creates a new avro writer provider

func NewCsvReader ¶

func NewCsvReader(args ...interface{}) (func(source interface{}, state data.Map) (interface{}, error), error)

func NewProtoReader ¶

func NewProtoReader(args ...interface{}) (func(source interface{}, state data.Map) (interface{}, error), error)

NewProtoWriter creates a new proto writer provider

func NewProtoWriter ¶

func NewProtoWriter(args ...interface{}) (func(source interface{}, state data.Map) (interface{}, error), error)

NewProtoWriter creates a new proto writer provider

func RegisterProviders ¶

func RegisterProviders(providers []*endly.UdfProvider) error

RegisterProviders register the supplied providers

func TransformWithUDF ¶

func TransformWithUDF(context *endly.Context, udfName, source string, payload interface{}) (interface{}, error)

TransformWithUDF transform payload with provided UDFs name.

func URLJoin ¶

func URLJoin(source interface{}, state data.Map) (interface{}, error)

URLJoin joins base URL and URI path

func URLPath ¶

func URLPath(source interface{}, state data.Map) (interface{}, error)

URLPath return path from URL

Types ¶

type ProtoCodec ¶

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

ProtoCodec represent a proto codec

func NewProtoCodec ¶

func NewProtoCodec(schemaFile, importPath string, msgType string, lowercaseKey bool) (*ProtoCodec, error)

NewProtoCodec creates a new protobuf codec

func (*ProtoCodec) AsBinary ¶

func (c *ProtoCodec) AsBinary(msgType string, msg interface{}) ([]byte, error)

func (*ProtoCodec) AsMessage ¶

func (c *ProtoCodec) AsMessage(msgType string, data []byte) (interface{}, error)

type RegisterRequest ¶

type RegisterRequest struct {
	UDFs []*endly.UdfProvider `description:"collection of predefined udf provider name with custom parameters and new registration id"`
}

RegisterRequest represents a register udf request

func NewRegisterRequestFromURL ¶

func NewRegisterRequestFromURL(URL string) (*RegisterRequest, error)

type RegisterResponse ¶

type RegisterResponse struct{}

RegisterRequest represents a register response

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
buildin

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL