Processing Filters

These filters are available for use with Kafka Connect File Pulse:

AppendFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.AppendFilter

The AppendFilter is probably one of the most important processing filters to know. It allows you to manipulate a source record by easily adding or replacing a field with a constant value or a value extracted from another existing field using ScEL.

Configuration

Examples

The following examples shows how to use the AppendFilter to concat two values from the array field named values using a substitution expression . The concat value is then added to the field named result.

Configuration

filters=SubstituteFilter
filters.SubstituteFilter.type=io.streamthoughts.kafka.connect.filepulse.filter.AppendFilter
filters.SubstituteFilter.field="$.result"
filters.SubstituteFilter.value="{{ extract_array($.values,0) }}-{{ extract_array($.values,1) }}"

Input

{
  "record": {
    "value": [
      "Hello",
      "World"
    ]
  }
} 

Output

{
  "record": {
    "value": [
      "Hello",
      "World"
    ],
    "result": "Hello-World"
  }
}

In the previous example, we used the simple property expression result to indicate the target field to which our value is added. We have actually omitted the expression scope $value. By default, if no scope is defined in an expression, the scope $value is implicitly applied. Hence, we could have used the fully expression $value.result which is similar to the simplified expression result.

But, you can perfectly use another expression scope. For example, you can leverage the AppendFilter to dynamically resolve the record-key or the output topic based on the record data.

The following configuration show how to use the $topic scope :

filters.SubstituteFilter.field="$topic"
filters.SubstituteFilter.value="my-topic-{{ lowercase(extract_array($.values,0)) }}"

Input

{
  "record": {
    "value": [
      "Hello",
      "World"
    ]
  }
} 

Output

{
  "context": {
    "topic": "my-topic-hello"
  },
  "record": {
    "value": [
      "Hello",
      "World"
    ]
  }
} 

Finally, the AppendFilter can also accept a substitution expression for the property field. This allows to dynamically determine the name of the field to be added.

The following examples show how to use a property expression to get the named of the field from a

filters.SubstituteFilter.field="$.target"
filters.SubstituteFilter.value="{{ extract_array($.values, 0) }}-{{ extract_array($.values,1) }}"

Input

{
  "record": {
    "target": "result",
    "value": [
      "Hello",
      "World"
    ]
  }
}

Output

{
  "record": {
    "target": "result",
    "value": [
      "Hello",
      "World"
    ],
    "result": "Hello-World"
  }
}

ConvertFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.ConvertFilter

The ConvertFilter can be used to convert a field’s value into a specific type.

Configuration

Supported types are :

• SHORT
• INTEGER
• LONG
• FLOAT
• DOUBLE
• BOOLEAN
• STRING
• ARRAY
• BYTES

Examples

The following example shows how to convert a a field’s value containing the string yes into a boolean.

Configuration

filters.BooleanConverter.field="target"
filters.BooleanConverter.to="BOOLEAN"

Input

{
  "record": {
    "target": "yes"
  }
}

Output

{
  "record": {
    "target": true
  }
}

CsvFilter

The following provides usage information for: io.streamthoughts.kafka.connect.filepulse.filter.CSVFilter

The CsvFilter can be used to parse a message field’s value containing columns delimited by a character into a struct

Configuration

Examples

The following example shows the use of the CSVFilter to split the message field using a comma (,) as a separator character. The name of each column is extracted from the fields headers.

filters=ParseCSVLine
filters.ParseCSVLine.type="io.streamthoughts.kafka.connect.filepulse.filter.CSVFilter"
filters.ParseCSVLine.extract.column.name="headers"
filters.ParseCSVLine.separator=","
filters.ParseCSVLine.trim.column="true"

DateFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.DateFilter

The DateFilter converts a field’s value containing a date to a unix epoch time.

Configuration

Examples

filters=ParseISODate
filters.ParseISODate.type=io.streamthoughts.kafka.connect.filepulse.filter.DateFilter
filters.ParseISODate.field="$.date"
filters.ParseISODate.target="$.timestamp"
filters.ParseISODate.formats="yyyy-MM-dd'T'HH:mm:ss"

Input

{
  "record": {
    "date": "2001-07-04T12:08:56"
  }
}

Output

{
  "record": {
    "date": "2001-07-04T12:08:56",
    "timestamp": 994248536000
  }
}

DelimitedRowFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.DelimitedRowFilter.

The DelimitedRowFilter can be used to parse and stream delimited row files (i.e CSV) into Kafka. Each row is parsed and published into a configured topic as a single Kafka data.

Configuration

Examples

The following example shows the use of the DelimitedRowFilter to split the message field using | as a separator character. The name of each column is extracted from the fields headers.

filters=ParseDelimitedRow
filters.ParseDelimitedRow.extractColumnNam="headers"
filters.ParseDelimitedRow.separator="\\|"
filters.ParseDelimitedRow.trimColumn="true"
filters.ParseDelimitedRow.type="io.streamthoughts.kafka.connect.filepulse.filter.DelimitedRowFilter"

DropFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.DropFilter.

The DropFilter can be used to prevent some messages (i.e records) to be written into Kafka.

Configuration

For more information about if property, see : Conditional execution.

Examples

The following example shows the usage of DropFilter to only keep records with a field level containing to ERROR.

filters=Drop
filters.Drop.type=io.streamthoughts.kafka.connect.filepulse.filter.DropFilter
filters.Drop.if={{ equals($.level, 'ERROR') }}
filters.Drop.invert=true

ExcludeFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.ExcludeFilter.

The ExcludeFilter can be used to exclude one or more fields from the input record.

Configuration

Examples

The following example shows the usage of ExcludeFilter.

filters=Exclude
filters.Exclude.type=io.streamthoughts.kafka.connect.filepulse.filter.ExcludeFilter
filters.Exclude.fields=message

Input

{
  "record": {
    "message": "{\"name\":\"pulse\"}",
    "name": "pulse"
  }
}

Output

{
  "record": {
    "name": "pulse"
  }
}

ExcludeFieldsMatchingPatternFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.ExcludeFieldsMatchingPatternFilter.

The ExcludeFieldsMatchingPatternFilter can be used to exclude all fields from the input record that are matching a regex pattern.

Configuration

Examples

The following example shows the usage of ExcludeFieldsMatchingPatternFilter.

filters=Exclude
filters.Exclude.type=io.streamthoughts.kafka.connect.filepulse.filter.ExcludeFieldsMatchingPatternFilter
filters.Exclude.regex=name.*

Input

{
  "record": {
    "message": "{\"name\":\"pulse\"}",
    "name1": "pulse",
    "name2": "pulse2"
  }
}

Output

{
  "record": {
    "message": "{\"name\":\"pulse\"}",
    "name1": null,
    "name2": null
  }
}

ExplodeFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.ExplodeFilter.

The ExplodeFilter can be used to explode an array or list field into separate records.

Configuration

Examples

The following example shows the usage of ExplodeFilter.

filters=Explode
filters.Explode.type=io.streamthoughts.kafka.connect.filepulse.filter.ExplodeFilter
filters.Explode.source=measurements

Input (single record)

{
  "record": {
    "id": "captor-0001",
    "date": "2020-08-06T17:00:00",
    "measurements": [
      38,
      40,
      42,
      37
    ]
  }
}

Output (multiple records)

{
  "record": {
    "id": "captor-0001",
    "date": "2020-08-06T17:00:00",
    "measurements": 38
  }
}
{
  "record": {
    "id": "captor-0001",
    "date": "2020-08-06T17:00:00",
    "measurements": 40
  }
}
{
  "record": {
    "id": "captor-0001",
    "date": "2020-08-06T17:00:00",
    "measurements": 42
  }
}
{
  "record": {
    "id": "captor-0001",
    "date": "2020-08-06T17:00:00",
    "measurements": 37
  }
}

ExtractValueFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.ExtractValueFilter.

The ExtractValueFilter can be used to extract the desired value out of a specific field.

Configuration

Examples

The following example shows the usage of ExtractValueFilter.

filters=Extract
filters.Extract.type=io.streamthoughts.kafka.connect.filepulse.filter.ExtractValueFilter
filters.Extract.field=name
filters.Extract.target=eventTime
filters.Extract.regex=(\d{4}-\d{2}-\d{2})

Input

{
  "record": {
    "name": "name_2023-08-28"
  }
}

Output

{
  "record": {
    "name": "name_2023-08-28",
    "eventTime": "2023-08-28"
  }
}

FailFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.FailFilter.

The fail filter can be used to throw an exception with a provided error message. For example, this can be useful to stop processing a file when a non-conform record is read.

Configuration

Examples

The following example shows the usage of FailFilter to stop processing a file when a field is equals to null.

filters=Fail
filters.Fail.type=io.streamthoughts.kafka.connect.filepulse.filter.FailFilter
filters.Fail.if={{ is_null($.user_id) }}
filters.Fail.message=Invalid row, user_id is missing : {{ $value }}

GrokFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.GrokFilter.

The GrokFilter allows you to parse unstructured data like applications logs to extract structured and meaningful data fields.

The GrokFilteris based on: https://github.com/streamthoughts/kafka-connect-transform-grok

Configuration

Examples

The following example shows the usage of GrokFilter to parse and extract fields from application log message.

filters=ParseLog4jLog
filters.ParseLog4jLog.pattern="%{TIMESTAMP_ISO8601:logdate} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}"
filters.ParseLog4jLog.overwrite="message"
filters.ParseLog4jLog.source="message"
filters.ParseLog4jLog.type="io.streamthoughts.kafka.connect.filepulse.filter.GrokFilter"
filters.ParseLog4jLog.ignoreFailure="true"

GroupRowFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.GroupRowFilter.

Configuration

Examples

JoinFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.JoinFilter.

Configuration

Examples

JSONFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.JSONFilter.

The JSONFilter parses an input json field.

Configuration

Examples

filters=MyJsonFilter
filters.MyJsonFilter.type=io.streamthoughts.kafka.connect.filepulse.filter.JSONFilter
filters.MyJsonFilter.source=message
filters.MyJsonFilter.target=payload

MultiRowFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.MultiRowFilter.

The MultiRowFilter joins multiple lines into a single Struct using a regex pattern.

Configuration

MoveAllFieldsFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.MoveAllFieldsFilter.

The MoveAllFieldsFilter moves all existing record field’s value to a specific target path.

Configuration

Examples

The following example shows the usage of the MoveAllFieldsFilter.

filters=MyMoveFilter
filters.MyMoveFilter.type=io.streamthoughts.kafka.connect.filepulse.filter.MoveAllFieldsFilter
filters.MyMoveFilter.target=payload
filters.MyMoveFilter.excludes=fieldC,fieldD

Input

{
  "record": {
    "fieldA": "foo",
    "fieldB": "bar",
    "fieldC": "foo",
    "fieldD": "bar"
  }
}

Output

{
  "record": {
    "fieldC": "foo",
    "fieldD": "bar",
    "payload": {
      "fieldA": "foo",
      "fieldB": "bar"
    }
  }
}

MoveFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.MoveFilter.

The MoveFilter moves an existing record field’s value to a specific target path.

Configuration

Examples

The following example shows the usage of the MoveFilter.

filters=MyMoveFilter
filters.MyMoveFilter.type=io.streamthoughts.kafka.connect.filepulse.filter.MoveFilter
filters.MyMoveFilter.source=field.child
filters.MyMoveFilter.target=moved

Input

{
  "record": {
    "field": {
      "child": "foo"
    }
  }
}

Output

{
  "record": {
    "moved": "foo"
  }
}

NamingConventionRenamefilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.NamingConventionRenameFilter.

The NamingConventionRenameFilter renames all fields in the struct to conform with a specific naming convention.

Configuration

Examples

filters=NormaliseFields
filters.NormaliseFields.type=io.streamthoughts.kafka.connect.filepulse.filter.NamingConventionRenameFilter
filters.NormaliseFields.naming.convention=camelCase
filters.NormaliseFields.delimiter=_

Input

{
  "record": {
    "RECORD_ID": "foo",
    "NAME": "bar"
  }
}

Output

{
  "record": {
    "recordId": "foo",
    "name": "bar"
  }
}

NullValueFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.NullValueFilter.

The NullValueFilter is used to empty a record-value to null.

Example

filters=NullValueIfDeleteOp
filters.NullValueIfDeleteOp.type=io.streamthoughts.kafka.connect.filepulse.filter.NullValueFilter
filters.NullValueIfDeleteOp.if={{ equals($value.op, 'DELETE') }}

RenameFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.RenameFilter.

The RenameFilter is used to rename a specified field.

Configuration

Examples

filters=RenameInputField
filters.RenameInputField.type=io.streamthoughts.kafka.connect.filepulse.filter.RenameFilter
filters.RenameInputField.field=input
filters.RenameInputField.target=renamed

Input

{
  "record": {
    "input": "foo"
  }
}

Output

{
  "record": {
    "renamed": "foo"
  }
}

SplitFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.SplitFilter.

The SplitFilter splits a field’s value of type string into an array by using a specific separator.

Configuration

Example

Configuration

filters=SplitInputField
filters.SplitInputField.type=io.streamthoughts.kafka.connect.filepulse.filter.SplitFilter
filters.SplitInputField.split=input
filters.SplitInputField.separator=,

Input

{
  "record": {
    "input": "val0,val1,val2"
  }
}

Output

{
  "record": {
    "input": "val0,val1,val2",
    "output": [
      "val0",
      "val1",
      "val2"
    ]
  }
}

XmlToJsonFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.XmlToJsonFilter.

The XmlToJsonFilter parses and converts an XML record-field it to a JSON string. This is filter is based on the org.json.XML library.

Configuration

Example

Configuration

filters=XmlToJson
filters.XmlToJson.type=io.streamthoughts.kafka.connect.filepulse.filter.XmlToJsonFilter
filters.XmlToJson.xml.parser.keep.strings=false
filters.XmlToJson.xml.parser.cDataTagName=data

XmlToStructFilter

The following provides usage information for : io.streamthoughts.kafka.connect.filepulse.filter.XmlToStructFilter.

The XmlToStructFilter parses an XML record-field into STRUCT

Configuration

Example

Configuration

filters=XmlToStruct
filters.ParseXmlDocument.type=io.streamthoughts.kafka.connect.filepulse.filter.XmlToStructFilter
filters.ParseXmlDocument.source=message
filters.ParseXmlDocument.xml.parser.validating.enabled=true
filters.ParseXmlDocument.xml.parser.namespace.aware.enabled=true
filters.ParseXmlDocument.xml.exclude.empty.elements=true
filters.ParseXmlDocument.xml.data.type.inference.enabled=true