Core Pipeline Components

package io.cdap.cdap.etl.api;

public abstract class Transform<IN, OUT> 
    implements StageLifecycle<TransformContext>, 
               SubmitterLifecycle<StageSubmitterContext>,
               Transformation<IN, OUT>, PipelineConfigurable {
    
    public static final String PLUGIN_TYPE = "transform";
    
    // Lifecycle methods
    public void configurePipeline(PipelineConfigurer pipelineConfigurer) {}
    public void initialize(TransformContext context) throws Exception {}
    public void prepareRun(StageSubmitterContext context) throws Exception {}
    public void onRunFinish(boolean succeeded, StageSubmitterContext context) {}
    public void destroy() {}
    
    // Access to runtime context
    protected TransformContext getContext();
}

@Plugin(type = Transform.PLUGIN_TYPE)
@Name("TextCleaner")
@Description("Cleans and normalizes text fields")
public class TextCleanerTransform extends Transform<StructuredRecord, StructuredRecord> {
    
    private final Config config;
    private Schema outputSchema;
    
    @Override
    public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
        StageConfigurer stageConfigurer = pipelineConfigurer.getStageConfigurer();
        Schema inputSchema = stageConfigurer.getInputSchema();
        
        if (inputSchema != null) {
            outputSchema = buildOutputSchema(inputSchema);
            stageConfigurer.setOutputSchema(outputSchema);
        }
        
        config.validate(stageConfigurer.getFailureCollector());
    }
    
    @Override
    public void initialize(TransformContext context) throws Exception {
        outputSchema = context.getOutputSchema();
    }
    
    @Override
    public void transform(StructuredRecord input, Emitter<StructuredRecord> emitter) 
            throws Exception {
        StructuredRecord.Builder builder = StructuredRecord.builder(outputSchema);
        
        for (Schema.Field field : input.getSchema().getFields()) {
            Object value = input.get(field.getName());
            if (field.getSchema().getType() == Schema.Type.STRING && value != null) {
                // Clean text: trim, normalize whitespace, remove special chars
                String cleanedValue = value.toString().trim().replaceAll("\\s+", " ");
                builder.set(field.getName(), cleanedValue);
            } else {
                builder.set(field.getName(), value);
            }
        }
        
        emitter.emit(builder.build());
    }
}

package io.cdap.cdap.etl.api;

public interface Transformation<IN, OUT> {
    /**
     * Transform input record and emit zero or more output records.
     * 
     * @param input input record to transform
     * @param emitter emitter for output records
     * @throws Exception if transformation fails
     */
    void transform(IN input, Emitter<OUT> emitter) throws Exception;
}

package io.cdap.cdap.etl.api;

public abstract class SplitterTransform<IN, OUT> 
    implements MultiOutputPipelineConfigurable,
               StageLifecycle<TransformContext>,
               SubmitterLifecycle<StageSubmitterContext>,
               MultiOutputTransformation<IN, OUT> {
    
    public static final String PLUGIN_TYPE = "splittertransform";
}

package io.cdap.cdap.etl.api;

public interface MultiOutputTransformation<IN, E> {
    /**
     * Transform input and emit to multiple named outputs.
     */
    void transform(IN input, MultiOutputEmitter<E> emitter) throws Exception;
}

public class DataSplitter extends SplitterTransform<StructuredRecord, StructuredRecord> {
    
    @Override
    public void transform(StructuredRecord input, MultiOutputEmitter<StructuredRecord> emitter) {
        String category = input.get("category");
        
        if ("valid".equals(category)) {
            emitter.emit("valid-data", input);
        } else if ("error".equals(category)) {
            emitter.emit("error-data", input);
        } else {
            emitter.emit("unknown-data", input);
        }
    }
}

package io.cdap.cdap.etl.api;

public abstract class ErrorTransform<ERR_IN, OUT> 
    implements StageLifecycle<TransformContext>,
               SubmitterLifecycle<StageSubmitterContext>,
               PipelineConfigurable {
    
    public static final String PLUGIN_TYPE = "errortransform";
    
    public void initialize(TransformContext context) throws Exception {}
    
    /**
     * Transform error record.
     */
    public abstract void transform(ErrorRecord<ERR_IN> input, Emitter<OUT> emitter) 
        throws Exception;
}

package io.cdap.cdap.etl.api;

public interface Emitter<T> extends AlertEmitter, ErrorEmitter<T> {
    /**
     * Emit a record to the next stage.
     */
    void emit(T value);
}

package io.cdap.cdap.etl.api;

public interface ErrorEmitter<T> {
    /**
     * Emit an error record.
     */
    void emitError(InvalidEntry<T> invalidEntry);
}

package io.cdap.cdap.etl.api;

public interface AlertEmitter {
    /**
     * Emit an alert with payload.
     */
    void emitAlert(Map<String, String> payload);
}

package io.cdap.cdap.etl.api;

public interface MultiOutputEmitter<E> extends AlertEmitter, ErrorEmitter<Object> {
    /**
     * Emit to a specific output port.
     */
    void emit(String port, E value);
}

package io.cdap.cdap.etl.api;

public interface Aggregator<GROUP_KEY, GROUP_VALUE, OUT> {
    /**
     * Emit group key for the group value.
     */
    void groupBy(GROUP_VALUE groupValue, Emitter<GROUP_KEY> emitter) throws Exception;
    
    /**
     * Aggregate all values for a group key into output records.
     */
    void aggregate(GROUP_KEY groupKey, Iterator<GROUP_VALUE> groupValues, 
                  Emitter<OUT> emitter) throws Exception;
}

public class SumAggregator implements Aggregator<String, StructuredRecord, StructuredRecord> {
    
    @Override
    public void groupBy(StructuredRecord groupValue, Emitter<String> emitter) throws Exception {
        String groupKey = groupValue.get("department");
        emitter.emit(groupKey);
    }
    
    @Override
    public void aggregate(String groupKey, Iterator<StructuredRecord> groupValues, 
                         Emitter<StructuredRecord> emitter) throws Exception {
        double sum = 0.0;
        int count = 0;
        
        while (groupValues.hasNext()) {
            StructuredRecord record = groupValues.next();
            Double salary = record.get("salary");
            if (salary != null) {
                sum += salary;
                count++;
            }
        }
        
        StructuredRecord result = StructuredRecord.builder(outputSchema)
            .set("department", groupKey)
            .set("total_salary", sum)
            .set("employee_count", count)
            .set("average_salary", count > 0 ? sum / count : 0.0)
            .build();
            
        emitter.emit(result);
    }
}

package io.cdap.cdap.etl.api;

public interface ReducibleAggregator<GROUP_KEY, GROUP_VALUE, AGGREGATE_VALUE, OUT> {
    /**
     * Emit group key for the group value.
     */
    void groupBy(GROUP_VALUE groupValue, Emitter<GROUP_KEY> emitter) throws Exception;
    
    /**
     * Aggregate group values into intermediate aggregate values.
     */
    void aggregate(GROUP_KEY groupKey, Iterator<GROUP_VALUE> groupValues, 
                  Emitter<AGGREGATE_VALUE> emitter) throws Exception;
    
    /**
     * Reduce intermediate aggregate values into final output.
     */
    void reduce(GROUP_KEY groupKey, Iterator<AGGREGATE_VALUE> aggregateValues, 
               Emitter<OUT> emitter) throws Exception;
}

package io.cdap.cdap.etl.api;

public interface Joiner<JOIN_KEY, INPUT_RECORD, OUT> {
    /**
     * Get join keys for input record (deprecated - use getJoinKeys).
     */
    @Deprecated
    default JOIN_KEY joinOn(String stageName, INPUT_RECORD inputRecord) throws Exception {
        throw new UnsupportedOperationException("joinOn method is deprecated");
    }
    
    /**
     * Get collection of join keys for input record.
     */
    default Collection<JOIN_KEY> getJoinKeys(String stageName, INPUT_RECORD inputRecord) 
        throws Exception {
        JOIN_KEY key = joinOn(stageName, inputRecord);
        return key == null ? Collections.emptySet() : Collections.singleton(key);
    }
    
    /**
     * Get join configuration.
     */
    JoinConfig getJoinConfig() throws Exception;
    
    /**
     * Merge records from join result.
     */
    OUT merge(JOIN_KEY joinKey, Iterable<JoinElement<INPUT_RECORD>> joinResult) 
        throws Exception;
}

package io.cdap.cdap.etl.api;

public class JoinConfig {
    /**
     * Create join config with required input stages.
     */
    public JoinConfig(Iterable<String> requiredInputs) {}
    
    /**
     * Get required input stages for the join.
     */
    public Iterable<String> getRequiredInputs() {}
}

package io.cdap.cdap.etl.api;

public class JoinElement<INPUT_RECORD> {
    public JoinElement(String stageName, INPUT_RECORD inputRecord) {}
    
    public String getStageName() {}
    public INPUT_RECORD getInputRecord() {}
}

package io.cdap.cdap.etl.api;

public interface StageLifecycle<T> extends Destroyable {
    /**
     * Initialize stage with runtime context.
     */
    void initialize(T context) throws Exception;
}

package io.cdap.cdap.etl.api;

public interface SubmitterLifecycle<T> {
    /**
     * Prepare for pipeline run.
     */
    void prepareRun(T context) throws Exception;
    
    /**
     * Handle run completion.
     */
    void onRunFinish(boolean succeeded, T context);
}

package io.cdap.cdap.etl.api;

public interface Destroyable {
    /**
     * Cleanup resources.
     */
    void destroy();
}

package io.cdap.cdap.etl.api;

public interface StageContext extends RuntimeContext, PluginContext, 
                                    ServiceDiscoverer, FeatureFlagsProvider,
                                    ConnectionConfigurable {
    // Provides access to:
    // - Runtime arguments and metrics
    // - Plugin instantiation
    // - Service discovery
    // - Feature flags
    // - Connection configuration
}

package io.cdap.cdap.etl.api;

public interface TransformContext extends StageContext, LookupProvider {
    // Inherits StageContext capabilities
    // Adds lookup provider for data lookups
}

package io.cdap.cdap.etl.api;

public interface StageSubmitterContext {
    /**
     * Get runtime arguments.
     */
    Arguments getArguments();
    
    /**
     * Get stage metrics.
     */
    StageMetrics getMetrics();
}

package io.cdap.cdap.etl.api;

public interface PipelineConfigurable {
    /**
     * Configure the pipeline stage.
     */
    void configurePipeline(PipelineConfigurer pipelineConfigurer);
}

package io.cdap.cdap.etl.api;

public interface MultiInputPipelineConfigurable {
    /**
     * Configure multi-input pipeline stage.
     */
    void configurePipeline(MultiInputPipelineConfigurer multiInputPipelineConfigurer);
}

package io.cdap.cdap.etl.api;

public interface MultiOutputPipelineConfigurable {
    /**
     * Configure multi-output pipeline stage.
     */
    void configurePipeline(MultiOutputPipelineConfigurer multiOutputPipelineConfigurer);
}

package io.cdap.cdap.etl.api;

public class ErrorRecord<T> {
    public ErrorRecord(T record, String errorMessage, int stage) {}
    public ErrorRecord(T record, String errorMessage) {}
    
    public T getRecord() {}
    public String getErrorMessage() {}
    public int getStage() {}
}

package io.cdap.cdap.etl.api;

public class InvalidEntry<T> {
    public InvalidEntry(int errorCode, String errorMsg, T invalidRecord) {}
    
    public int getErrorCode() {}
    public String getErrorMsg() {}
    public T getInvalidRecord() {}
}

package io.cdap.cdap.etl.api;

public interface SerializableTransform<IN, OUT> 
    extends Transformation<IN, OUT>, Serializable {
    // Combines transformation with Java serialization
}

package io.cdap.cdap.etl.api;

public interface ToKeyValueTransform<IN, KEY, VAL> {
    /**
     * Transform input into key-value pairs.
     */
    void transform(IN input, Emitter<KeyValue<KEY, VAL>> emitter) throws Exception;
}

package io.cdap.cdap.etl.api;

public interface FromKeyValueTransform<KEY, VAL, OUT> {
    /**
     * Transform key-value pairs into output records.
     */
    void transform(KeyValue<KEY, VAL> input, Emitter<OUT> emitter) throws Exception;
}