Status

Current state: WIP Accepted

Discussion thread: here

JIRA: KAFKA-3209

Released: 0.10.2.0

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

This proposal is for adding a record transformation API to Kafka Connect as well as certain bundled transformations. At the same time, we should not extend Connect's area of focus too much beyond the Extract and Load aspects of ETLbeyond moving data between Kafka and other systems. We will only support simple 1:{0,1} transformations – i.e. map and filter operations.

...

• Allow for lightweight updates to records.
  • Some transformations must be performed before the data hits Kafka (source) or another system (sink) e.g. filtering certain types of events or sensitive information.
  • It's also useful for very light , single-message modifications that are easier to perform inline with the data import/export. It may be inconvenient to add stream processing into the mix for simple data massaging or control over routing.

...

// Existing base class for SourceRecord and SinkRecord, new self type parameter.
public abstract class ConnectRecord<R extends ConnectRecord<R>> {
 
    // ...
 
    // New abstract method:
       
    /** Generate a new record of the same type as itself, with the specified parameter values. **/
    public abstract R newRecord(String topic, Schema keySchema, Object key, Schema valueSchema, Object value, Long timestamp);
 
}

public interface Transformation<R extends ConnectRecord<R>> extends Configurable, Closeable {
 
    //** via InitializeConfigurable withbase theinterface:
 provided configuration properties. **//
    void initconfigure(Map<String, String>?> configconfigs);
 
    /**
     * Apply transformation to the {@code record} and return another record object (which may be {@code record} itself) or {@code null},
     * corresponding to a map or filter operation respectively. The implementation Mustmust be thread-safe.
     **/
    R apply(R record);
 
    /** Configuration Signalspecification thatfor this transformation instance will no longer will be used. **/
    voidConfigDef closeconfig();
 
    /** ConfigurationSignal specification forthat this transformation instance will no longer will be used. **/
    @Override
    ConfigDefvoid configclose();
 
}

Configuration

A transformation chain will be configured at the connector-level. The order of transformations is defined by the transforms config which represents a list of aliases. An alias in transforms implies that some additional keys are configurable:
- transforms.$alias.type – fully qualified class name for the transformation
- transforms.$alias.* – all other keys as defined in Transformation.config() are embedded with this prefix

...

If the result of any Transformation.apply() in a chain is null, that record is discarded (not written to Kafka in the case of a source connector, or not provided to sink connector).

Bundled transformations

Criteria: SMTs that are shipped with Kafka Connect should be general enough to apply to many data sources & serialization formats. They should also be simple enough to not cause any additional library dependency to be introduced.

Beyond those being initially included with this KIP, transformations can be adopted for inclusion in future with JIRA/ML discussion to weigh the tradeoffsWIP. Subject to discussion –  not a final list.

ValueToKey

Hoist{Key,Value}ToStruct

Extract{Key,Value}FromStruct

Patterns for implementing data transformations

• Data transformations could be applicable to the key or the value of the record. We will have *Key and *Value variants for these transformations that reuse the common functionality from a shared base class.

• Some common utilities for data transformations will shape up:

  • Cache the changes they make to Schema objects, possibly only preserving last-seen one as the likelihood of source data Schema changing is low.

  • Copying of Schema objects with the possible exclusion of some fields, which they are modifying. Likewise, copying of Struct object to another Struct having a different Schema with the exception of some fields, which they are modifying.

  • Where fields are being added and a field name specified in configuration, we will want a consistent way to convey if it should be created as an a required or optional field. We can use a leading '!' or '?' character . TODO: specify escapingfor this purpose if the user wants to make a different choice than the default determined by the transformation.
    

  • ConfigDef does not provide a Type.MAP, but for the time being we can piggyback on top of Type.LIST and represent maps as a list of key-value pairs separated by : TODO: specify escaping.
  • Where field names are expected, in some cases we should allow for getting at nested fields by allowing a dotted syntax which is common in such usage (and accordingly, will need some utilities around accessing a field that may be nested). TODO: specify escaping
  • There are escaping considerations to several such configs, so we will need utilities that that assume a consistent escaping style (e.g. backslashes).

Compatibility, Deprecation, and Migration Plan

...

However, the surface area for such a change is much larger - we would need additional REST APIs for creating, updating and validating transformation chain configs. The current proposal does not prevent taking this direction down the line.

Not including any transformations with Connect

In the interest of providing a better out-of-the-box experience and avoiding duplication of effort in the ecosystem, we will be bundling certain transformations with Connect.

One concern here is that we should have a well-defined criteria for what belongs in Connect vs external dependencies, which was addressed.