Filter (Confluent)
The following provides usage information for the Confluent SMT
io.confluent.connect.transforms.Filter
. Include or drop records that match the filter.condition
Predicate.
Description
Include or drop records that match the filter.condition
predicate.
The filter.condition
is a predicate specifying JSON Path that is applied to each record processed, and when this predicate successfully matches the record is either included (when filter.type=include
) or excluded (when filter.type=exclude
).
The missing.or.null.behavior
property defines how the transform behaves when a record does not have the field(s) used in the filter condition predicate. By default the behavior is to fail. This property can also be set to include or exclude the record that is missing the predicate’s field(s).
Use the transformation type designed for the record key
(io.confluent.connect.transforms.Filter$Key
) or value
(io.confluent.connect.transforms.Filter$Value
).
Installation
This transformation is developed by Confluent and does not ship by default with
Kafka or Confluent Platform. You can install this transformation using the Confluent Hub Client:
confluent-hub install confluentinc/connect-transforms:latest
Examples
The configuration snippet below shows how to use and configure the Confluent Filter SMT.
The filter.condition
is a predicate specifying JSON path. For more
information on JSON path format, see https://github.com/json-path/JsonPath.
filter.condition
specified in the examples below is satisfied when the
nested field 'nestedKey'
in field 'key'
is equal to either 'value1'
or 'value2'
.
filterExample1
defines the configuration for including
records satisfying the filter condition.
The filter condition shall be applied on record value
. The fail
behavior in configuration specifies to throw an exception and fail the connector task
when the record does not have the field(s) used in the filter condition.
transforms=filterExample1
# Use the 'io.confluent.connect.transforms.Filter$Value' as the source (record 'value') on which
# 'filter.condition' shall be applied.
transforms.filterExample1.type=io.confluent.connect.transforms.Filter$Value
transforms.filterExample1.filter.condition=$.key[?(@.nestedKey == "value1" || @nestedKey == "value2")]
# Use 'include' to pass through all records that match the predicate.
transforms.filterExample1.filter.type=include
# Use the 'fail' behavior to throw an exception and fail the connector task when the record does
# not have the field(s) used in the 'filter.condition'.
transforms.filterExample1.missing.or.null.behavior=fail
filterExample2
defines the configuration for excluding
records satisfying the filter condition.
The filter condition shall be applied on record key
. The include
behavior in configuration specifies to pass the record through
when the record does not have the field(s) used in the filter condition.
transforms=filterExample2
# Use the 'io.confluent.connect.transforms.Filter$Key' as the source (record 'key') on which
# 'filter.condition' shall be applied.
transforms.filterExample2.type=io.confluent.connect.transforms.Filter$Key
transforms.filterExample2.filter.condition=$.key[?(@.nestedKey == "value1" || @nestedKey == "value2")]
# Use 'exclude' to drop all records that match the predicate.
transforms.filterExample2.filter.type=exclude
# Use the 'include' behavior to pass the record through when the record does not have the field(s)
# used in the 'filter.condition'.
transforms.filterExample2.missing.or.null.behavior=include
Properties
Name |
Description |
Type |
Default |
Valid Values |
Importance |
filter.condition |
Specifies the criteria used to match records to be included or excluded
by this transformation. Use JSON Path predicate notation defined in:
https://github.com/json-path/JsonPath. |
string |
|
|
high |
filter.type |
Specifies the action to perform with records that match the
filter.condition predicate. Use include to pass through all
records that match the predicate and drop all records that do not
satisfy the predicate, or use exclude to drop all records that match
the predicate. |
string |
|
[include, exclude] |
high |
missing.or.null.behavior |
Specifies the behavior when the record does not have the field(s) used
in the filter.condition . Use fail to throw an exception and fail
the connector task, include to pass the record through, or
exclude to drop the record. |
string |
fail |
[fail, include, exclude] |
medium |