# Kafka Connect JMESPath: Expressive content-based record filtering

**URL:** <https://forum.confluent.io/t/kafka-connect-jmespath-expressive-content-based-record-filtering/1104>\
**Category:** Kafka Connect\
**Created:** [3 March 2021 21:08 UTC](https://forum.confluent.io/t/kafka-connect-jmespath-expressive-content-based-record-filtering/1104 "2021-03-03T21:08:25Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![denisw](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/denisw/32/577_2.png) [@denisw](https://forum.confluent.io/u/denisw)\
**Post date:** [3 March 2021 21:08 UTC](https://forum.confluent.io/t/kafka-connect-jmespath-expressive-content-based-record-filtering/1104/1 "2021-03-03T21:08:26Z")

</div>

Hi,

This week, I published the first version of a Kafka Connect plugin that solves a need I often had: filtering records based on the contents of its key or value.

The existing _Filter (Confluent)_ transform seems to fill that need, but its choice of JSONPath as the query language for conditions makes it very limited. Most severely, JSONPath’s conditional syntax is limited to arrays, so it is not possible to express basic conditions like “the value of field `foo` must equal `"bar"`”.

This is why I created [Kafka Connect JMESPath](https://www.confluent.io/hub/denisw/kafka-connect-jmespath). This plugin integrates the JMESPath query language, which is way more powerful than JSONPath and makes it straightforward to express both simple and complex conditions, such as:

- `genre == 'Mystery'`
- `author == 'Agatha Christie' && publishDate.year >= `1950` `
- `contains(title, 'Murder')`.

Another difference that the JMESPath plugin does not define another type of transform. Instead, it is based on Kafka Connect’s general transform predicate system, which, while most commonly combined with the [Filter (Apache Kafka)](https://docs.confluent.io/platform/current/connect/transforms/filter-ak.html#predicates) transform for record filtering, can be used with _any_ transform to apply it only conditionally. This makes the JMESPath predicates more widely applicable than the Confluent Filter transform.

**Example Usage:**

```json
{
  "connector.class": "FileStreamSink",
  "topics": "books",
  "file": "/tmp/books.txt",
  "tasks.max": "1",

  "transforms": "Filter",
  "transforms.Filter.type": "org.apache.kafka.connect.transforms.Filter",
  "transforms.Filter.predicate": "Old",

  "predicates": "Old",
  "predicates.Old.type": "de.denisw.kafka.connect.jmespath.MatchesJMESPath$Value",
  "predicates.Old.query": "publishDate.year < `2000`"
}

```

The JMESPath plugin is open-source (Apache License 2.0) and available on GitHub. The README contains the full documentation on how to use the transform predicates. I hope it will be useful to some of you. Feedback welcome!

---

<div class="post-metadata">

**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\
**Post date:** [3 March 2021 22:06 UTC](https://forum.confluent.io/t/kafka-connect-jmespath-expressive-content-based-record-filtering/1104/2 "2021-03-03T22:06:45Z")

</div>

Sounds great, thanks for sharing @denisw. Looks like I’ll need to go and revise [🎄 Twelve Days of SMT 🎄 - Day 11: Predicate and Filter](https://rmoff.net/2020/12/22/twelve-days-of-smt-day-11-predicate-and-filter/) to include it 🙂

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [17 March 2021 22:07 UTC](https://forum.confluent.io/t/kafka-connect-jmespath-expressive-content-based-record-filtering/1104/3 "2021-03-17T22:07:16Z")

</div>

This topic was automatically closed 14 days after the last reply. New replies are no longer allowed.