Make schema validation optional

examples/fraud/resources/weight_column.yaml

@@ -5,18 +5,9 @@
     columns:
       col: class
 
-- kind: transformer
-  name: weight
-  inputs:
-    columns:
-      col: INT_COLUMN
-    args:
-      class_distribution: {INT: FLOAT}
-  output_type: FLOAT_COLUMN
-
 - kind: transformed_column
   name: weight_column
-  transformer: weight
+  transformer_path: implementations/transformers/weight.py
   inputs:
     columns:
       col: class

examples/mnist/resources/transformed_columns.yaml

@@ -1,6 +1,6 @@
 - kind: transformed_column
   name: image_pixels
-  transformer: decode_and_normalize
+  transformer_path: implementations/transformers/decode_and_normalize.py
   inputs:
     columns:
       image: image

examples/reviews/resources/max_length.yaml

@@ -1,13 +1,6 @@
-- kind: aggregator
-  name: max_length
-  inputs:
-    columns:
-      col: STRING_COLUMN
-  output_type: INT
-
 - kind: aggregate
   name: max_review_length
-  aggregator: max_length
+  aggregator_path: implementations/aggregators/max_length.py
   inputs:
     columns:
       col: review

examples/reviews/resources/tokenized_columns.yaml

@@ -1,16 +1,6 @@
-- kind: transformer
-  name: tokenize_string_to_int
-  output_type: INT_LIST_COLUMN
-  inputs:
-    columns:
-      col: STRING_COLUMN
-    args:
-      max_len: INT
-      vocab: {STRING: INT}
-
 - kind: transformed_column
   name: embedding_input
-  transformer: tokenize_string_to_int
+  transformer_path: implementations/transformers/tokenize_string_to_int.py
   inputs:
     columns:
       col: review

pkg/consts/consts.go

@@ -36,21 +36,24 @@ var (
 	RequirementsTxt = "requirements.txt"
 	PackageDir      = "packages"
 
-	AppsDir             = "apps"
-	DataDir             = "data"
-	RawDataDir          = "data_raw"
-	TrainingDataDir     = "data_training"
-	AggregatorsDir      = "aggregators"
-	AggregatesDir       = "aggregates"
-	TransformersDir     = "transformers"
-	ModelImplsDir       = "model_implementations"
-	PythonPackagesDir   = "python_packages"
-	ModelsDir           = "models"
-	ConstantsDir        = "constants"
-	ContextsDir         = "contexts"
-	ResourceStatusesDir = "resource_statuses"
-	WorkloadSpecsDir    = "workload_specs"
-	LogPrefixesDir      = "log_prefixes"
+	AppsDir               = "apps"
+	APIsDir               = "apis"
+	DataDir               = "data"
+	RawDataDir            = "data_raw"
+	TrainingDataDir       = "data_training"
+	AggregatorsDir        = "aggregators"
+	AggregatesDir         = "aggregates"
+	TransformersDir       = "transformers"
+	ModelImplsDir         = "model_implementations"
+	PythonPackagesDir     = "python_packages"
+	ModelsDir             = "models"
+	ConstantsDir          = "constants"
+	ContextsDir           = "contexts"
+	ResourceStatusesDir   = "resource_statuses"
+	WorkloadSpecsDir      = "workload_specs"
+	LogPrefixesDir        = "log_prefixes"
+	RawColumnsDir         = "raw_columns"
+	TransformedColumnsDir = "transformed_columns"
 
 	TelemetryURL = "https://telemetry.cortexlabs.dev"
 )

pkg/operator/api/context/aggregators.go

@@ -25,8 +25,9 @@ type Aggregators map[string]*Aggregator
 type Aggregator struct {
 	*userconfig.Aggregator
 	*ResourceFields
-	Namespace *string `json:"namespace"`
-	ImplKey   string  `json:"impl_key"`
+	Namespace      *string `json:"namespace"`
+	ImplKey        string  `json:"impl_key"`
+	SkipValidation bool    `json:"skip_validation"`
 }
 
 func (aggregators Aggregators) OneByID(id string) *Aggregator {

pkg/operator/api/context/context.go

@@ -55,6 +55,7 @@ type Resource interface {
 	GetID() string
 	GetIDWithTags() string
 	GetResourceFields() *ResourceFields
+	GetMetadataKey() string
 }
 
 type ComputedResource interface {
@@ -72,6 +73,7 @@ type ResourceFields struct {
 	ID           string        `json:"id"`
 	IDWithTags   string        `json:"id_with_tags"`
 	ResourceType resource.Type `json:"resource_type"`
+	MetadataKey  string        `json:"metadata_key"`
 }
 
 type ComputedResourceFields struct {
@@ -91,6 +93,10 @@ func (r *ResourceFields) GetResourceFields() *ResourceFields {
 	return r
 }
 
+func (r *ResourceFields) GetMetadataKey() string {
+	return r.MetadataKey
+}
+
 func (r *ComputedResourceFields) GetWorkloadID() string {
 	return r.WorkloadID
 }

pkg/operator/api/context/models.go

@@ -38,12 +38,11 @@ type Model struct {
 }
 
 type TrainingDataset struct {
-	userconfig.ResourceConfigFields
+	userconfig.ResourceFields
 	*ComputedResourceFields
-	ModelName   string `json:"model_name"`
-	TrainKey    string `json:"train_key"`
-	EvalKey     string `json:"eval_key"`
-	MetadataKey string `json:"metadata_key"`
+	ModelName string `json:"model_name"`
+	TrainKey  string `json:"train_key"`
+	EvalKey   string `json:"eval_key"`
 }
 
 func (trainingDataset *TrainingDataset) GetResourceType() resource.Type {

pkg/operator/api/context/python_packages.go

@@ -24,7 +24,7 @@ import (
 type PythonPackages map[string]*PythonPackage
 
 type PythonPackage struct {
-	userconfig.ResourceConfigFields
+	userconfig.ResourceFields
 	*ComputedResourceFields
 	SrcKey     string `json:"src_key"`
 	PackageKey string `json:"package_key"`

pkg/operator/api/context/transformers.go

@@ -25,8 +25,9 @@ type Transformers map[string]*Transformer
 type Transformer struct {
 	*userconfig.Transformer
 	*ResourceFields
-	Namespace *string `json:"namespace"`
-	ImplKey   string  `json:"impl_key"`
+	Namespace      *string `json:"namespace"`
+	ImplKey        string  `json:"impl_key"`
+	SkipValidation bool    `json:"skip_validation"`
 }
 
 func (transformers Transformers) OneByID(id string) *Transformer {

pkg/operator/api/userconfig/aggregates.go

@@ -26,11 +26,12 @@ import (
 type Aggregates []*Aggregate
 
 type Aggregate struct {
-	ResourceConfigFields
-	Aggregator string        `json:"aggregator" yaml:"aggregator"`
-	Inputs     *Inputs       `json:"inputs" yaml:"inputs"`
-	Compute    *SparkCompute `json:"compute" yaml:"compute"`
-	Tags       Tags          `json:"tags" yaml:"tags"`
+	ResourceFields
+	Aggregator     string        `json:"aggregator" yaml:"aggregator"`
+	AggregatorPath *string       `json:"aggregator_path" yaml:"aggregator_path"`
+	Inputs         *Inputs       `json:"inputs" yaml:"inputs"`
+	Compute        *SparkCompute `json:"compute" yaml:"compute"`
+	Tags           Tags          `json:"tags" yaml:"tags"`
 }
 
 var aggregateValidation = &configreader.StructValidation{
@@ -45,10 +46,13 @@ var aggregateValidation = &configreader.StructValidation{
 		{
 			StructField: "Aggregator",
 			StringValidation: &configreader.StringValidation{
-				Required:                      true,
-				AlphaNumericDashDotUnderscore: true,
+				AllowEmpty: true,
 			},
 		},
+		{
+			StructField:         "AggregatorPath",
+			StringPtrValidation: &configreader.StringPtrValidation{},
+		},
 		inputValuesFieldValidation,
 		sparkComputeFieldValidation("Compute"),
 		tagsFieldValidation,

pkg/operator/api/userconfig/aggregators.go

@@ -24,7 +24,7 @@ import (
 type Aggregators []*Aggregator
 
 type Aggregator struct {
-	ResourceConfigFields
+	ResourceFields
 	Inputs     *Inputs     `json:"inputs"  yaml:"inputs"`
 	OutputType interface{} `json:"output_type"  yaml:"output_type"`
 	Path       string      `json:"path"  yaml:"path"`

pkg/operator/api/userconfig/apis.go

@@ -24,7 +24,7 @@ import (
 type APIs []*API
 
 type API struct {
-	ResourceConfigFields
+	ResourceFields
 	ModelName string      `json:"model_name" yaml:"model_name"`
 	Compute   *APICompute `json:"compute" yaml:"compute"`
 	Tags      Tags        `json:"tags" yaml:"tags"`

pkg/operator/api/userconfig/config.go

@@ -200,19 +200,47 @@ func (config *Config) Validate(envName string) error {
 		}
 	}
 
-	// Check local aggregators exist
+	// Check local aggregators exist or a path to one is defined
 	aggregatorNames := config.Aggregators.Names()
 	for _, aggregate := range config.Aggregates {
-		if !strings.Contains(aggregate.Aggregator, ".") && !slices.HasString(aggregatorNames, aggregate.Aggregator) {
-			return errors.Wrap(ErrorUndefinedResource(aggregate.Aggregator, resource.AggregatorType), Identify(aggregate), AggregatorKey)
+		if aggregate.AggregatorPath == nil && aggregate.Aggregator == "" {
+			return ErrorMissingAggregator(aggregate)
+		}
+
+		if aggregate.AggregatorPath != nil && aggregate.Aggregator != "" {
+			return ErrorMultipleAggregatorSpecified(aggregate)
+		}
+
+		switch {
+		case aggregate.AggregatorPath != nil:
+			continue
+		case aggregate.Aggregator != "":
+			if !strings.Contains(aggregate.Aggregator, ".") &&
+				!slices.HasString(aggregatorNames, aggregate.Aggregator) {
+				return errors.Wrap(ErrorUndefinedResource(aggregate.Aggregator, resource.AggregatorType), Identify(aggregate), AggregatorKey)
+			}
 		}
 	}
 
-	// Check local transformers exist
+	// Check local transformers exist or a path to one is defined
 	transformerNames := config.Transformers.Names()
 	for _, transformedColumn := range config.TransformedColumns {
-		if !strings.Contains(transformedColumn.Transformer, ".") && !slices.HasString(transformerNames, transformedColumn.Transformer) {
-			return errors.Wrap(ErrorUndefinedResource(transformedColumn.Transformer, resource.TransformerType), Identify(transformedColumn), TransformerKey)
+		if transformedColumn.TransformerPath == nil && transformedColumn.Transformer == "" {
+			return ErrorMissingTransformer(transformedColumn)
+		}
+
+		if transformedColumn.TransformerPath != nil && transformedColumn.Transformer != "" {
+			return ErrorMultipleTransformerSpecified(transformedColumn)
+		}
+
+		switch {
+		case transformedColumn.TransformerPath != nil:
+			continue
+		case transformedColumn.Transformer != "":
+			if !strings.Contains(transformedColumn.Transformer, ".") &&
+				!slices.HasString(transformerNames, transformedColumn.Transformer) {
+				return errors.Wrap(ErrorUndefinedResource(transformedColumn.Transformer, resource.TransformerType), Identify(transformedColumn), TransformerKey)
+			}
 		}
 	}
 

pkg/operator/api/userconfig/constants.go

@@ -25,7 +25,7 @@ import (
 type Constants []*Constant
 
 type Constant struct {
-	ResourceConfigFields
+	ResourceFields
 	Type  interface{} `json:"type" yaml:"type"`
 	Value interface{} `json:"value" yaml:"value"`
 	Tags  Tags        `json:"tags" yaml:"tags"`

pkg/operator/api/userconfig/embed.go

@@ -24,7 +24,7 @@ import (
 type Embeds []*Embed
 
 type Embed struct {
-	ResourceConfigFields
+	ResourceFields
 	Template string                 `json:"template" yaml:"template"`
 	Args     map[string]interface{} `json:"args" yaml:"args"`
 }

pkg/operator/api/userconfig/environments.go

@@ -28,7 +28,7 @@ import (
 type Environments []*Environment
 
 type Environment struct {
-	ResourceConfigFields
+	ResourceFields
 	LogLevel *LogLevel `json:"log_level" yaml:"log_level"`
 	Limit    *Limit    `json:"limit" yaml:"limit"`
 	Data     Data      `json:"-" yaml:"-"`

pkg/operator/api/userconfig/errors.go

@@ -58,6 +58,10 @@ const (
 	ErrK8sQuantityMustBeInt
 	ErrRegressionTargetType
 	ErrClassificationTargetType
+	ErrMissingAggregator
+	ErrMissingTransformer
+	ErrMultipleAggregatorSpecified
+	ErrMultipleTransformerSpecified
 )
 
 var errorKinds = []string{
@@ -90,9 +94,13 @@ var errorKinds = []string{
 	"err_k8s_quantity_must_be_int",
 	"err_regression_target_type",
 	"err_classification_target_type",
+	"err_missing_aggregator",
+	"err_missing_transformer",
+	"err_multiple_aggregator_specified",
+	"err_multiple_transformer_specified",
 }
 
-var _ = [1]int{}[int(ErrClassificationTargetType)-(len(errorKinds)-1)] // Ensure list length matches
+var _ = [1]int{}[int(ErrMultipleTransformerSpecified)-(len(errorKinds)-1)] // Ensure list length matches
 
 func (t ErrorKind) String() string {
 	return errorKinds[t]
@@ -376,9 +384,38 @@ func ErrorRegressionTargetType() error {
 		message: "regression models can only predict float target values",
 	}
 }
+
 func ErrorClassificationTargetType() error {
 	return Error{
 		Kind:    ErrClassificationTargetType,
 		message: "classification models can only predict integer target values (i.e. {0, 1, ..., num_classes-1})",
 	}
 }
+
+func ErrorMissingAggregator(aggregate *Aggregate) error {
+	return Error{
+		Kind:    ErrMissingAggregator,
+		message: fmt.Sprintf("missing aggregator for aggregate \"%s\", expecting either \"aggregator\" or \"aggregator_path\"", aggregate.Name),
+	}
+}
+
+func ErrorMissingTransformer(transformedColumn *TransformedColumn) error {
+	return Error{
+		Kind:    ErrMissingTransformer,
+		message: fmt.Sprintf("missing transformer for transformed_column \"%s\", expecting either \"transformer\" or \"transformer_path\"", transformedColumn.Name),
+	}
+}
+
+func ErrorMultipleAggregatorSpecified(aggregate *Aggregate) error {
+	return Error{
+		Kind:    ErrMultipleAggregatorSpecified,
+		message: fmt.Sprintf("aggregate \"%s\" specified both \"aggregator\" and \"aggregator_path\", please specify only one", aggregate.Name),
+	}
+}
+
+func ErrorMultipleTransformerSpecified(transformedColumn *TransformedColumn) error {
+	return Error{
+		Kind:    ErrMultipleTransformerSpecified,
+		message: fmt.Sprintf("transformed_column \"%s\" specified both  \"transformer\" and \"transformer_path\", please specify only one", transformedColumn.Name),
+	}
+}

pkg/operator/api/userconfig/models.go

@@ -29,7 +29,7 @@ import (
 type Models []*Model
 
 type Model struct {
-	ResourceConfigFields
+	ResourceFields
 	Type               ModelType                `json:"type" yaml:"type"`
 	Path               string                   `json:"path" yaml:"path"`
 	TargetColumn       string                   `json:"target_column" yaml:"target_column"`