Quick Flow: Consume from Kafka and filter data

Learn about this Quick Flow that consumes from a Kafka topic applies a SQL filter, and sends the filtered data to a Kafka topic in JSON format. Review flow details, download the template, and understand the template parameters.

About this Quick Flow

This flow definition consumes from a Kafka topic, applies SQL Filter to filter the stream and sends filtered stream to a destination Kafka topic in JSON format. The user must configure a Schema Registry connection and specify a schema for the source Kafka topic which allows the user to provide SQL to filter the stream. If the events from the source Kafka topic do not adhere to the schema, the event is logged and not sent to the destination Kafka topic.

Quick Flow definition details and download

Quick Flow definition details
Flow Name Kafka_CSV-Filter-to-Kafka_JSON_SchemaRegistry
Flow Artifact Download the file:


Source Kafka Topic
Source Format CSV
Destination Kafka Topic
Destination Format JSON

Quick Flow parameters

Parameter Name Description Sensitive Example
CDP Env Truststore File Provide the Truststore file that you generated for the environment where your Kafka cluster is running in N
CDP Env Trustore Password Provide the password for the Truststore you created for the environment where your Kafka cluster is running in Y
CDP Schema Registry URL Specify the Schema Registry URL you want to connect to in the format https://hostname:7790/api/v1 N https://dataflow-streams-master0.​dataflow.xcu2-8y8x.dev.cldr.work:​7790/api/v1
CSV Delimiter Specify the delimiter for the source CSV data N , ; |
Filter Rule Specify the filter rule expressed in SQL to filter streaming events for the destination source N SELECT * FROM FLOWFILE where id=4
Kafka Broker Endpoint Specify the Kafka bootstrap servers string as a comma separated list N messaging-cluster-1-broker0.gvettica.​xcu2-8y8x.dev.cldr.work:9093,​messaging-cluster-1-broker1.​gvettica.xcu2-8y8x.dev.cldr.work:9093,​messaging-cluster-1-broker2.gvettica.​xcu2-8y8x.dev.cldr.work:9093
Kafka Consumer Group ID Specify the id for the consumer group used for the the source topic you are consuming from N
Kafka Destination Topic Specify the destination topic name N
Kafka Producer ID Specify the id for the Kafka producer writing into the destination topic N
Kafka Service Account

Specify the CDP machine user name that you want to use to authenticate to Kafka. Ensure this user has the appropriate access rights to the Kafka topics.

Kafka Service Account Password Specify the password of the CDP machine user you're using to authenticate against Kafka Y
Kafka Source Topic Specify a topic name or list of topics with the same schema that you want to read from N
Schema Name Specify the schema name to be looked up in the Schema Registry N