DEV Community: Florian Hussonnois

Streaming data into Kafka S01/E04 — Parsing log files using Grok Expressions

Florian Hussonnois — Tue, 05 Jan 2021 14:36:21 +0000

This is the fourth and last article of the series "Streaming data into Kafka" series. In the first three articles, we saw how it's fairly easy to use Kafka Connect to load records from CSV, XML and JSON files into Apache Kafka without writing a single line of code. For doing this, we have used the Kafka Connect FilePulse connector which packs with a lot of nice features to parse and transform data.

In this last article, we are going to see how to parse unstructured logs files from an NGINX web server into structured data fields.

Kafka Connect File Pulse connector

If you have already read the previous articles go directly to the next section (i.e Ingesting Data).

The Kafka Connect FilePulse connector is a powerful source connector that makes it easy to parse, transform, and load data from the local file system into Apache Kafka. It offers built-in support for various file formats (e.g: CSV, XML, JSON, LOG4J, AVRO).

For a broad overview of FilePulse, I suggest you read this article:

Kafka Connect FilePulse - One Connector to Ingest them All!

For more information, you can check-out the documentation here.

How to use the connector

The easiest and fastest way to get started with the Kafka Connect FilePulse connector is to use the Docker image available on Docker Hub.

$ docker pull streamthoughts/kafka-connect-file-pulse:1.6.3

You can download the docker-compose.yml file available on the GitHub project repository to quickly start a Confluent Platform with Kafka Connect and the FilePulse connector pre-installed.

$ wget https://raw.githubusercontent.com/streamthoughts/kafka-connect-file-pulse/v1.6.3/docker-compose.yml
$ docker-compose up -d

Once all Docker containers are started, you can check that the connector is installed on the Kafka Connect worker accessible on http://localhost:8083.

$ curl -s localhost:8083/connector-plugins|jq '.[].class'|egrep FilePulse

"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector"

Note: You can also install the connector either from GitHub Releases Page or from Confluent Hub.

Starting NGINX

First, let's start an NGINX instance to serve a single HTML page, using Docker.

Let's create some directories for the sake of this demo.

$ mkdir -p demo/content demo/logs

...and a simple HTML page called ìndex.html:

$ cat <<EOF > demo/content/index.html
<!DOCTYPE html>
<html>
  <head>
    <title>Hi!</title>
  </head>
  <body>
    <h1>Hello World - Kafka Connect FilePulse</h1>
    <strong>You can add a Star to this repository to support us! Thank You<a href="https://github.com/streamthoughts/kafka-connect-file-pulse">GitHub</a></strong>
  </body>
</html>
EOF

Then, start the NGINX web server, by running:

$ docker run --name nginx \
-p 8080:80 \
-v `pwd`/demo/content:/usr/share/nginx/html:ro -d nginx

Check that the server is running properly:

$ curl -X GET http://localhost:8080

Finally, and to simplify things for the rest of the article, we will redirect stderr and stdout of the container to the ./demo/logs/nginx.log file.

$ docker logs -f nginx > ./demo/logs/nginx.log 2>&1 &

Ingesting Data

First, let's stop the container running Kafka Connect that was started using docker-compose.

$ docker stop connect && docker rm connect

Then, start a new one with a mounted volume for accessing the nginx.log file.

Create a file to define the environment variables that must be set for Kafka Connect container.

$ cat <<EOF > connect-file-pulse-env.list
CONNECT_BOOTSTRAP_SERVERS=localhost:9092
CONNECT_REST_ADVERTISED_HOST_NAME=connect
CONNECT_REST_PORT=8083
CONNECT_GROUP_ID=compose-connect-group
CONNECT_CONFIG_STORAGE_TOPIC=docker-connect-configs
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=1
CONNECT_OFFSET_FLUSH_INTERVAL_MS=10000
CONNECT_OFFSET_STORAGE_TOPIC=docker-connect-offsets
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=1
CONNECT_STATUS_STORAGE_TOPIC=docker-connect-status
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=1
CONNECT_KEY_CONVERTER=org.apache.kafka.connect.storage.StringConverter
CONNECT_VALUE_CONVERTER=io.confluent.connect.avro.AvroConverter
CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL=http://localhost:8081
CONNECT_INTERNAL_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter
CONNECT_INTERNAL_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter
CONNECT_ZOOKEEPER_CONNECT=localhost:2181
CONNECT_PLUGIN_PATH=/usr/share/java,/usr/share/confluent-hub-components/
EOF

Start the container running the Kafka Connect FilePulse connector:

$ docker run -it \
 --network=host \
 --name=connect \
 --env-file connect-file-pulse-env.list \
 -v `pwd`/demo/logs:/tmp/connect-data \
 streamthoughts/kafka-connect-file-pulse:latest

Then, create a new connector with the below configuration:

$ curl \
    -sX PUT -H "Accept:application/json" \
    -H "Content-Type:application/json" http://localhost:8083/connectors/source-log-filepulse-01/config \
    -d '{
     "connector.class": "io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
     "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
     "fs.scanner.class": "io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker",
     "fs.scan.directory.path": "/tmp/connect-data",
     "fs.scan.interval.ms": "10000",
     "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
     "file.filter.regex.pattern":".*\\.log$",
     "internal.kafka.reporter.bootstrap.servers": "localhost:9092",
     "internal.kafka.reporter.topic": "connect-file-pulse-status",
     "offset.strategy": "name",
     "read.max.wait.ms": "900000",
     "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
     "topic": "connect-file-pulse-nginx-raw-logs",
     "tasks.max": 1
}' | jq

Finally, you can consume the Topic named connect-file-pulse-nginx-raw-logs to verify that the connector has detected the .log file:

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t connect-file-pulse-nginx-raw-logs \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "message": {
    "string": "172.17.0.1 - - [05/Jan/2021:10:56:52 +0000] \"GET / HTTP/1.1\" 200 306 \"-\" \"curl/7.58.0\" \"-\""
  }
}

Eventually, you can generate more access logs by running:

$ for i in $(seq 0 100); do curl -sX GET http://localhost:8080 >/dev/null; sleep 1 ; done &

Note: In the example above, we have used kafkacat to consume the topics. The option -o-1 is used to only consume the latest message

Ok, let's take a moment to describe the configuration we've just used.

First, the connector will periodically scan the input directory /tmp/connect-data that we set through the property fs.scan.directory.path and will lookup for files matching the pattern .*\\.log$.

Then, each file will be uniquely identified and tracked depending on the value of the offset.strategy property. Here, the configuration specifies that a file must be identified by its name. Alternatively, we could, for example, have chosen to use the inode of the file as an identifier. Connect FilePulse supports multiple identifiers that can be combined (e.g: name+hash).

In addition, the connector is configured to use the RowFileInputReader (see: task.reader.class) that allows creating one Kafka record per line.

One of the characteristics of the RowFileInputReader is that will not immediately complete the processing after hitting the end of the file but will wait until reaching a timeout for more bytes to be written to the file. This behavior is configured through the read.max.wait.ms property. Here we are waiting 15 minutes before finishing the file processing.

Parsing Data using Grok Expressions

So far, we have been able to read the NGINX logs continuously. Each time a new line is added to the file a new record is sent to Kafka containing a single text field called message. But, it would be preferable to be able to parse each line to extract useful data and to produce structured messages into Kafka.

The Elastic/ELK Stack and in particular the Logstash solution has popularised the use of Grok expressions to parse and transform unstructured data into meaningful fields. Grok sits on top of Regular Expression (regex) to match relevant data using text patterns.

Connect FilePulse brings the power of Grok Expression directly to Kafka Connect with the GrokFilter that under the hood uses the Joni library the Java port of Oniguruma regexp library. It also provides a lot of predefined and reusable grok patterns. See the complete list of patterns.

Let's define a custom grok pattern to match lines of the NGINX access log file.

$ cat <<EOF > nginx
NGINX_ACCESS %{IPORHOST:remote_addr} - %{USERNAME:remote_user} \[%{HTTPDATE:time_local}\] \"%{DATA:request}\" %{INT:status} %{NUMBER:bytes_sent} \"%{DATA:http_referer}\" \"%{DATA:http_user_agent}\"
EOF

Note: The syntax for a grok pattern is %{SYNTAX:SEMANTIC} or %{SYNTAX:SEMANTIC:TYPE}.

Then, we have to make the pattern available to the connector by copying the nginx file, previously created, into the container :

$ docker exec -it connect mkdir -p /tmp/grok-patterns
$ docker cp nginx connect://tmp/grok-patterns/nginx

After that, we can create a new connector with the following configuration :

$ curl \
    -sX PUT -H "Accept:application/json" \
    -H "Content-Type:application/json" http://localhost:8083/connectors/source-log-filepulse-02/config \
    -d '{
     "connector.class": "io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
     "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
     "fs.scanner.class": "io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker",
     "fs.scan.directory.path": "/tmp/connect-data",
     "fs.scan.interval.ms": "10000",
     "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
     "file.filter.regex.pattern":".*\\.log$",
     "internal.kafka.reporter.bootstrap.servers": "localhost:9092",
     "internal.kafka.reporter.topic": "connect-file-pulse-status",
     "offset.strategy": "name",
     "read.max.wait.ms": "120000",
     "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
     "topic": "connect-file-pulse-nginx-parsed-logs",
     "tasks.max": 1,
     "filters": "ParseLog",
     "filters.ParseLog.type": "io.streamthoughts.kafka.connect.filepulse.filter.GrokFilter",
     "filters.ParseLog.match": "%{NGINX_ACCESS}",
     "filters.ParseLog.overwrite": "message",
     "filters.ParseLog.source": "message",
     "filters.ParseLog.ignoreFailure": "true",
     "filters.ParseLog.patternsDir": "/tmp/grok-patterns"
}' | jq

Finally, let's consume the output Topic connect-file-pulse-nginx-parsed-logs to observe the extracted fields.

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t connect-file-pulse-nginx-parsed-logs \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "message": {
    "string": "172.17.0.1 - - [05/Jan/2021:13:14:54 +0000] \"GET / HTTP/1.1\" 200 306 \"-\" \"curl/7.58.0\" \"-\""
  },
  "remote_addr": {
    "string": "172.17.0.1"
  },
  "remote_user": {
    "string": "-"
  },
  "time_local": {
    "string": "05/Jan/2021:13:14:54 +0000"
  },
  "request": {
    "string": "GET / HTTP/1.1"
  },
  "status": {
    "string": "200"
  },
  "bytes_sent": {
    "string": "306"
  },
  "http_referer": {
    "string": "-"
  },
  "http_user_agent": {
    "string": "curl/7.58.0"
  }
}

As previously, you can Eventually generate more access logs by running:

$ for i in $(seq 0 100); do curl -sX GET http://localhost:8080 >/dev/null; sleep 1 ; done

Et voilà, It's as simple as that!

Using Grok Expression with Kafka Connect SMT

In the previous example, we use the processing filter chain feature provided by Connect FilePulse.

But, Kafka Connect already ships with a mechanism called Single Message Transforms (SMTs) that was added in Apache Kafka 0.10 (KIP-66). SMTs can be used to modify the data of each record that flow through Kafka Connect pipeline.

Good news ! We have externalized the work done with the GrokFilter to a dedicated SMT called: Kafka Connect Grok Transformation.

Conclusion

We have seen in this article that it is fairly easy to continuously read and parse log files. The Connect File Pulse connector ships with the GrokFilter to parse unstructured data using Grok expressions as you would have done with Logstash.

More generally, Connect File Pulse connector is a powerful solution that allows you to easily manipulate your data before sending it into Apache Kafka. Please do not hesitate to share this article. If you like this project then add a ⭐ to the GitHub repository to support us. Thank you.

Streaming data into Kafka S01/E03 - Loading JSON file

Florian Hussonnois — Thu, 10 Sep 2020 12:36:00 +0000

This is the third article in the "Streaming data into Kafka" series. In the first two, we saw how it's fairly easy to use Kafka Connect to load records from CSV and XML files into Apache Kafka without writing a single line of code. For doing this, we have used the Kafka Connect FilePulse connector which packs with a lot of nice features to parse and transform data.

Now, let's see how to integrate JSON data, another file format that is widely used on most projects (and much more appreciated than XML for web-based applications).

Kafka Connect File Pulse connector

If you have already read the previous articles go directly to the next section (i.e Ingesting Data).

For a broad overview of FilePulse, I suggest you read this article :

Kafka Connect FilePulse - One Connector to Ingest them All!

For more information, you can check-out the documentation here.

How to use the connector

The easiest and fastest way to get started with the Kafka Connect FilePulse connector is to use the Docker image available on Docker Hub.

$ docker pull streamthoughts/kafka-connect-file-pulse:1.6.3

You can download the docker-compose.yml file available on the GitHub project repository to quickly start a Confluent Platform with Kafka Connect and the FilePulse connector pre-installed.

$ wget https://raw.githubusercontent.com/streamthoughts/kafka-connect-file-pulse/v1.6.3/docker-compose.yml
$ docker-compose up -d

Once all Docker containers are started, you can check that the connector is installed on the Kafka Connect worker accessible on http://localhost:8083.

$ curl -s localhost:8083/connector-plugins|jq '.[].class'|egrep FilePulse

"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector"

Note : You can also install the connector either from GitHub Releases Page or from Confluent Hub.

Ingesting Data

To read a file containing a single JSON document, we will use the BytesArrayInputReader. This reader allows us to create a single record per source file. Each record produced by this reader will have a single field of type byte[] named message. The byte[] value is the full content of the source file (i.e the JSON document).

Then, to parse this field, we will use the processing filter mechanism provided by the FilePulse connector and more particularly the JSONFilter.

So lets's create the connector with this minimal configuration:

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/tracks-json-filepulse-00/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.json$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.BytesArrayInputReader",
        "offset.strategy":"name",
        "topic":"tracks-filepulse-json-00",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.DeleteCleanupPolicy",
        "filters": "ParseJSON",
        "filters.ParseJSON.type":"io.streamthoughts.kafka.connect.filepulse.filter.JSONFilter",
        "filters.ParseJSON.source":"message",
        "filters.ParseJSON.merge":"true",
        "tasks.max": 1
    }'

Note: The Connect FilePulse connector periodically scan the input directory that we set using the property fs.scan.directory.path. Then, it will lookup for files matching the pattern .*\\.json$. Each file is uniquely identified and tracked depending on the value of the offset.strategy. Here, the configuration specifies that a file is identified by its name.

Create a valid JSON file that looks like this:

$ cat <<EOF > track.json
{ 
  "track": {
     "title":"Star Wars (Main Theme)",
     "artist":"John Williams, London Symphony Orchestra",
     "album":"Star Wars",
     "duration":"10:52"
  }
}
EOF

Then copy this file from your host to the Docker container which runs the connector. You can run the following commands:

// Create the target directory
$ docker exec -it connect mkdir -p /tmp/kafka-connect/examples

// Copy host file to docker-container
$ docker cp track.json connect://tmp/kafka-connect/examples/track-00.json

Finally, consume the topic named tracks-filepulse-json-00 and verify that the connector has detected and processed the JSON file:

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t tracks-filepulse-json-00 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "message": {
    "bytes": "{ \n  \"track\": {\n     \"title\":\"Star Wars (Main Theme)\",\n     \"artist\":\"John Williams, London Symphony Orchestra\",\n     \"album\":\"Star Wars\",\n     \"duration\":\"10:52\"\n  }\n}\n"
  },
  "track": {
    "Track": {
      "title": {
        "string": "Star Wars (Main Theme)"
      },
      "artist": {
        "string": "John Williams, London Symphony Orchestra"
      },
      "album": {
        "string": "Star Wars"
      },
      "duration": {
        "string": "10:52"
      }
    }
  }
}

Note: In the example above, we have used kafkacat to consume the topics. The option -o-1 is used to only consume the latest message

Excluding Field

The JSONFilter does not automatically delete the original field containing the raw JSON string (i.e. the message). If you do not want to keep this field, you can remove it using the ExcludeFilter as follows:

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/tracks-json-filepulse-00/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.json$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.BytesArrayInputReader",
        "offset.strategy":"name",
        "topic":"tracks-filepulse-json-01",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.DeleteCleanupPolicy",
        "filters": "ParseJSON, ExcludeFieldMessage",
        "filters.ParseJSON.type":"io.streamthoughts.kafka.connect.filepulse.filter.JSONFilter",
        "filters.ParseJSON.source":"message",
        "filters.ParseJSON.merge":"true",
        "filters.ExcludeFieldMessage.type":"io.streamthoughts.kafka.connect.filepulse.filter.ExcludeFilter",
        "filters.ExcludeFieldMessage.fields":"message",
        "tasks.max": 1
    }'

Copy the JSON file to the Docker container as previously:

$ docker cp track.json \
connect://tmp/kafka-connect/examples/track-01.json

Then consume the output topic tracks-filepulse-json-01 by running :

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t tracks-filepulse-json-01 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "track": {
    "Track": {
      "title": {
        "string": "Star Wars (Main Theme)"
      },
      "artist": {
        "string": "John Williams, London Symphony Orchestra"
      },
      "album": {
        "string": "Star Wars"
      },
      "duration": {
        "string": "10:52"
      }
    }
  }
}

That's it! We have successfully produced a clean structured message similar to the one contained in our input file.

Now, let's go a step further.

Handling Null values

Sometimes you may have to process JSON documents with null values. By default, if we take the configuration used so far, the null values will be ignored during the serialization.

The main reason for this is that the connector cannot infer the type of a field containing a null value.

However, we can combine the AppendFilter and the Simple Connect Expression Language (SCEL) to both define the type of null value and set a default value.

Note: Simple Connect Expression Language (SCEL) is a basic expression language provided by the Connect FilePulse connector to access and manipulate records fields.

Let's update the connector configuration :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/tracks-json-filepulse-02/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.json$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.BytesArrayInputReader",
        "offset.strategy":"name",
        "topic":"tracks-filepulse-json-02",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.DeleteCleanupPolicy",
        "filters": "ParseJSON, ExcludeFieldMessage, SetDefaultRank",
        "filters.ParseJSON.type":"io.streamthoughts.kafka.connect.filepulse.filter.JSONFilter",
        "filters.ParseJSON.source":"message",
        "filters.ParseJSON.merge":"true",
        "filters.ExcludeFieldMessage.type":"io.streamthoughts.kafka.connect.filepulse.filter.ExcludeFilter",
        "filters.ExcludeFieldMessage.fields":"message",
        "filters.SetDefaultRank.type":"io.streamthoughts.kafka.connect.filepulse.filter.AppendFilter",
        "filters.SetDefaultRank.field":"$value.track.rank",
        "filters.SetDefaultRank.value":"{{ converts(nlv($value.track.rank, 0), '\''INTEGER'\'') }}",
        "filters.SetDefaultRank.overwrite": "true",
        "tasks.max": 1
    }'

Create a second JSON document with the following content :

$ cat <<EOF > track-with-null.json
{ 
  "track": {
     "title":"Duel of the Fates",
     "artist":"John Williams, London Symphony Orchestra",
     "album":"Star Wars",
     "duration":"4:14",
     "rank": null
  }
}
EOF

Copy it to the Docker container as previously:

$ docker cp track-with-null.json \
connect://tmp/kafka-connect/examples/track-02.json

Next consume the output topic tracks-filepulse-json-01 :

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t tracks-filepulse-json-02 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "track": {
    "Track": {
      "title": {
        "string": "Duel of the Fates"
      },
      "artist": {
        "string": "John Williams, London Symphony Orchestra"
      },
      "album": {
        "string": "Star Wars"
      },
      "duration": {
        "string": "4:14"
      },
      "rank": {
        "int": 0
      }
    }
  }
}

Finally, you should get an output message containing the field rank of type int and initialize with the default value 0.

Spliting JSON Array

Finally, it's also common to have to process JSON files containing a JSON array of records.

To produce one record per element in the array you have to set explode.array property of the JSONFilter to true.

Lets' update the connector configuration with the following :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/tracks-json-filepulse-00/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.json$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.BytesArrayInputReader",
        "offset.strategy":"name",
        "topic":"tracks-filepulse-json-03",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.DeleteCleanupPolicy",
        "filters": "ParseJSON, ExcludeFieldMessage",
        "filters.ParseJSON.type":"io.streamthoughts.kafka.connect.filepulse.filter.JSONFilter",
        "filters.ParseJSON.source":"message",
        "filters.ParseJSON.merge":"true",
        "filters.ParseJSON.explode.array":"true",
        "filters.ExcludeFieldMessage.type":"io.streamthoughts.kafka.connect.filepulse.filter.ExcludeFilter",
        "filters.ExcludeFieldMessage.fields":"message",
        "tasks.max": 1
    }'

Create a file containing two JSON objects :

$ cat <<EOF > tracks.json
[
  {
    "track": {
      "title": "Star Wars (Main Theme)",
      "artist":  "John Williams, London Symphony Orchestra",
      "album": "Star Wars",
      "duration": "10:52"
    }
  },
  {
    "track": {
      "title":  "Duel of the Fates",
      "artist": "John Williams, London Symphony Orchestra",
      "album": "Star Wars: The Phantom Menace (Original Motion Picture Soundtrack)",
      "duration": "4:14"
    }
  }
]
EOF

Copy it to the Docker container as previously.

$ docker cp tracks.json \
connect://tmp/kafka-connect/examples/tracks-00.json

Then consume the output topic tracks-filepulse-json-02 :

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t tracks-filepulse-json-03 \
        -C -J -q -o0 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "track": {
    "Track": {
      "title": {
        "string": "Star Wars (Main Theme)"
      },
      "artist": {
        "string": "John Williams, London Symphony Orchestra"
      },
      "album": {
        "string": "Star Wars"
      },
      "duration": {
        "string": "10:52"
      }
    }
  }
}
{
  "track": {
    "Track": {
      "title": {
        "string": "Duel of the Fates"
      },
      "artist": {
        "string": "John Williams, London Symphony Orchestra"
      },
      "album": {
        "string": "Star Wars: The Phantom Menace (Original Motion Picture Soundtrack)"
      },
      "duration": {
        "string": "4:14"
      }
    }
  }
}

And voilà! Now you know how to process JSON files with Kafka Connect.

Conclusion

We have seen in this article that it is very easy to load records from JSON files into Apache Kafka without writing a single line of code using Kafka Connect. The Connect File Pulse connector is a very powerful solution that allows you to easily manipulate your data before loading it into Apache Kafka.

Please, share this article if you like this project. You can even add a ⭐ to the GitHub repository to support us.

Thank you very much.

Streaming data into Kafka S01/E02 - Loading XML file

Florian Hussonnois — Wed, 19 Aug 2020 11:26:06 +0000

In the previous blog post Streaming data into Kafka S01/E01- Loading CSV file, I've illustrated how it can be easy to integrate data into Apache Kafka using the Kafka Connect framework.

In particular, we saw how to parse and transform CSV files to produce clean records into Kafka by using the Kafka Connect FilePulse connector.

XML(Extensible Markup Language) is another well-known data format. Usually, the XML format is not very appreciated by most developers because of its heaviness (or complexity). However, it's still used by many organizations to make systems interact with each other.

In this second article, we will see how to read records from XML files and load them into Kafka. To do this, we will once again use the Kafka Connect FilePulse connector, which offers native support for reading XML files.

Kafka Connect File Pulse connector

For a broad overview of FilePulse, I suggest you read this article :

For more information, you can check-out the documentation here.

How to use the connector

The easiest and fastest way to get started with the Kafka Connect FilePulse connector is to use the Docker image available on Docker Hub.

$ docker pull streamthoughts/kafka-connect-file-pulse:1.6.3

You can download the docker-compose.yml file available on the GitHub project repository to quickly start a Confluent Platform with Kafka Connect and the FilePulse connector pre-installed.

$ wget https://raw.githubusercontent.com/streamthoughts/kafka-connect-file-pulse/v1.6.3/docker-compose.yml
$ docker-compose up -d

Once all Docker containers are started, you can check that the connector is installed on the Kafka Connect worker accessible on http://localhost:8083.

$ curl -s localhost:8083/connector-plugins|jq '.[].class'|egrep FilePulse

"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector"

Note : You can also install the connector either from GitHub Releases Page or from Confluent Hub.

Ingesting Data

We will start by creating a first connector with the following configuration. It specifies that the connector's tasks must use the XMLFileInputReader to read the files that will be scheduled by the connector.

The connector will periodically scan the input directory that we set using the property fs.scan.directory.path. Then, it will lookup for files matching the pattern .*\\.xml$. Each file is uniquely identified and tracked depending on the value of the offset.strategy. Here, the configuration specifies that a file is identified by its name.

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-xml-filepulse-00/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.xml$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.XMLFileInputReader",
        "offset.strategy":"name",
        "topic":"playlists-filepulse-xml-00",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1
    }'

Once the connector is created, you can check that it is properly started by executing:

$ curl -s \
localhost:8083/connectors/source-xml-filepulse-00/status \
| jq  '.connector.state'

"RUNNING"

Now, let's create an XML files called playlists.xml with the following content :

$ cat <<EOF > playlists.xml
<?xml version="1.0" encoding="UTF-8"?>
<playlists>
    <playlist name="BestOfStarWars">
        <track>
            <title>Duel of the Fates</title>
            <artist>John Williams, London Symphony Orchestra</artist>
            <album>Star Wars: The Phantom Menace (Original Motion Picture Soundtrack)</album>
            <duration>4:14</duration>   
        </track>
        <track>
            <title>Star Wars (Main Theme)</title>
            <artist>John Williams, London Symphony Orchestra</artist>
            <album>Star Wars: The Empire Strikes Back (Original Motion Picture Soundtrack)</album>
            <duration>10:52</duration>  
        </track>
    </playlist> 
</playlists>
EOF

Then, copy this file from your host to the Docker container which runs the connector. You can run the following commands:

// Copy CSV file to docker-container
$ docker exec -it connect mkdir -p /tmp/kafka-connect/examples

$ docker cp playlists.xml connect://tmp/kafka-connect/examples/playlists-00.xml

Finally, you can consume the playlists-filepulse-xml-00 topic to verify that the connector has detected and processed the XML file:

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t playlists-filepulse-xml-00 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

{
  "playlists": {
    "Playlists": {
      "playlist": {
        "Playlist": {
          "name": {
            "string": "BestOfStarWars"
          },
          "track": {
            "array": [
              {
                "Track": {
                  "title": {
                    "string": "Duel of the Fates"
                  },
                  "artist": {
                    "string": "John Williams, London Symphony Orchestra"
                  },
                  "album": {
                    "string": "Star Wars: The Phantom Menace (Original Motion Picture Soundtrack)"
                  },
                  "duration": {
                    "string": "4:14"
                  }
                }
              },
              {
                "Track": {
                  "title": {
                    "string": "Star Wars (Main Theme)"
                  },
                  "artist": {
                    "string": "John Williams, London Symphony Orchestra"
                  },
                  "album": {
                    "string": "Star Wars: The Empire Strikes Back (Original Motion Picture Soundtrack)"
                  },
                  "duration": {
                    "string": "10:52"
                  }
                }
              }
            ]
          }
        }
      }
    }
  }
}

Note: In the example above, we are using kafkacat to consume messages. The option -o-1 is used to only consume the latest message.

As you have noticed from the above example, the Connect FilePulse has automatically inferred the schema from the input XML file.

So, let's check the produced Avro schema by using the HTTP SchemaRegistry endpoint :

$ curl -X GET \
http://localhost:8081/subjects/playlists-filepulse-xml-00-value/versions/latest/schema \
| jq

Forcing Array-type

Let's create a second XML file named single-track-playlist.xml with the following content :

$ cat <<EOF > single-track-playlists.xml
<?xml version="1.0" encoding="UTF-8"?>
<playlists>
    <playlist name="BestOfJWilliams">
        <track>
            <title>Theme From Jurassic Park</title>
            <artist>John Williams, London Symphony Orchestra</artist>
            <album>Jurassic Park</album>
            <duration>3:27</duration>   
        </track>
        </playlist>
</playlists>
EOF

Then copy it to the Docker container as previously.

$ docker cp single-track-playlists.xml \
connect://tmp/kafka-connect/examples/playlists-01.xml

Now, let's consume the topic playlists-filepulse-xml-00. You will see that the field named track is of type Track, whereas in the first example it was of type array.

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t playlists-filepulse-xml-00 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault
{
  "playlists": {
    "Playlists": {
      "playlist": {
        "Playlist": {
          "name": {
            "string": "BestOfJWilliams"
          },
          "track": {
            "Track": {
              "title": {
                "string": "Theme From Jurassic Park"
              },
              "artist": {
                "string": "John Williams, London Symphony Orchestra"
              },
              "album": {
                "string": "Jurassic Park"
              },
              "duration": {
                "string": "3:27"
              }
            }
          }
        }
      }
    }
  }
}

This is a common problem when trying to infer a data schema from an XML file. It's a bit difficult to identify that a field must be an array when only one element is present.

To solve this problem, it's possible to configure the XMLFileInputReader to force some fields to be of type array.

Let's update the connector with the given configuration :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-xml-filepulse-00/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.xml$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.XMLFileInputReader",
        "force.array.on.fields": "track",
        "offset.strategy":"name",
        "topic":"playlists-filepulse-xml-00",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1
    }'

Then recopy the XML file with a new name, so that it will be pickup by the connector.

$ docker cp single-track-playlists.xml connect://tmp/kafka-connect/examples/playlist-03.xml

Finally, consume the output topic to observe the effect of the new configuration.

docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \                       
        -b localhost:9092 \
        -t playlists-filepulse-xml-00 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "playlists": {
    "Playlists": {
      "playlist": {
        "Playlist": {
          "name": {
            "string": "BestOfJWilliams"
          },
          "track": {
            "array": [
              {
                "Track": {
                  "title": {
                    "string": "Theme From Jurassic Park"
                  },
                  "artist": {
                    "string": "John Williams, London Symphony Orchestra"
                  },
                  "album": {
                    "string": "Jurassic Park"
                  },
                  "duration": {
                    "string": "3:27"
                  }
                }
              }
            ]
          }
        }
      }
    }
  }
}

The track field is now an array field.

Splitting XML using XPath expression

So far we have only used the "XMLFileInputReader" to read a single record from each XML file.

But we could just as well produce one record for each track element present in the input XML files. To do this, we will simply override the property xpath.expression, whose default value is /.

So, let's create a new connector with the following configuration :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-xml-filepulse-01/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.xml$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.XMLFileInputReader",
        "xpath.expression": "//playlist/track",
        "offset.strategy":"name",
        "topic":"tracks-filepulse-xml-00",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1
    }'

Then, consume the messages from the output topic tracks-filepulse-xml-00

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t tracks-filepulse-xml-00 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "title": {
    "string": "Theme From Jurassic Park"
  },
  "artist": {
    "string": "John Williams, London Symphony Orchestra"
  },
  "album": {
    "string": "Jurassic Park"
  },
  "duration": {
    "string": "3:27"
  }
}

You should get one record per track.

Extracting string values

By default, the XMLFileInputReader expects the XPath expression to return a result of type NodeSet.

If your goal is to extract a single text element from an input XML files you will have to set the property xpath.result.type=STRING.

For example, if we configure a connector with the following XPath expression //playlist/track/title/text() then the connector will produce messages in the form :

{
  "message": {
    "string": "Theme From Jurassic Park"
  }
}

Then, you can use Single Message Transformations (SMT) such as ExtractField to replace the entire value with the extracted field.

Renaming fields

It's sometimes useful to be able to change the name of a field. This can be because you need to further contextualize a field or because you are not satisfied with the field names present in the XML input file.

For example, we can rename the field artist because the field contains a comma-separated list of artists (not a single one).

The File Pulse connector allows us to define complex pipelines to parse, transform, and enrich data through the use of processing Filters.

Thus, to rename the field artist into artists we will use the RenameFilter.

Let's create a new connector with this new configuration :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-xml-filepulse-02/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.xml$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.XMLFileInputReader",
        "force.array.on.fields": "track",
        "xpath.expression": "//playlist/track",
        "filters": "RenameArtist",
        "filters.RenameArtist.type": "io.streamthoughts.kafka.connect.filepulse.filter.RenameFilter",
        "filters.RenameArtist.field": "artist",
        "filters.RenameArtist.target": "artists",
        "offset.strategy":"name",
        "topic":"tracks-filepulse-xml-02",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1
    }'

Next, consume the produced messages from the output topic tracks-filepulse-xml-02

docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t tracks-filepulse-xml-02 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "title": {
    "string": "Theme From Jurassic Park"
  },
  "album": {
    "string": "Jurassic Park"
  },
  "duration": {
    "string": "3:27"
  },
  "artists": {
    "string": "John Williams, London Symphony Orchestra"
  }
}

Splitting fields into an array

Finally, I'm going to show you how you can combine the filters to build a complete processing pipeline.

For this last example we will split the "artists" field into a table using the SplitFilter.

Let's create a final connector with the following configuration :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-xml-filepulse-03/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.xml$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.XMLFileInputReader",
        "force.array.on.fields": "track",
        "xpath.expression": "//playlist/track",
        "filters": "RenameArtists, SplitArtists",
        "filters.RenameArtists.type": "io.streamthoughts.kafka.connect.filepulse.filter.RenameFilter",
        "filters.RenameArtists.field": "artist",
        "filters.RenameArtists.target": "artists",
        "filters.SplitArtists.type": "io.streamthoughts.kafka.connect.filepulse.filter.SplitFilter",
        "filters.SplitArtists.split": "artists",
        "filters.SplitArtists.separator": ",",
        "offset.strategy":"name",
        "topic":"tracks-filepulse-xml-03",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1
    }'

Finally, consume the output topic tracks-filepulse-xml-03 to observe the result of our filter-chain.

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t tracks-filepulse-xml-03 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

(output)

{
  "title": {
    "string": "Theme From Jurassic Park"
  },
  "album": {
    "string": "Jurassic Park"
  },
  "duration": {
    "string": "3:27"
  },
  "artists": {
    "array": [
      {
        "string": "John Williams"
      },
      {
        "string": " London Symphony Orchestra"
      }
    ]
  }
}

And voila!

Conclusion

We have seen in this article that it can be fairly easy to load records from XML files into Apache Kafka without writing a single line of code using Kafka Connect. Also, the Connect File Pulse connector is a powerful solution that allows you to easily manipulate your data before loading it into Apache Kafka.

Please, share this article if you like this project. You can even add a ⭐ to the GitHub repository to support us.

Thank you very much.

Streaming data into Kafka S01/E01 - Loading CSV file

Florian Hussonnois — Wed, 12 Aug 2020 17:52:07 +0000

Ingesting data files in Apache Kafka is a very common task. Among all the various file formats that we can find, CSV is probably the most popular one to move data between different systems. This is due to its simplicity and to the fact that it can be used to export or import data from one (small) database to another.

A CSV file is nothing more than a text file (with a .csv extension). Each line of the file represents a data record and each record consists of one or more fields, separated by a comma (or another separator).

Here is a chunk of example :

40;War;02:38;1983;U2;Rock
Acrobat;Achtung Baby;04:30;1991;U2;Rock
Sunday Bloody Sunday;War;04:39;1983;U2;Rock
With or Without You;The Joshua Tree;04:55;1987; U2;Rock

In Addition, a CSV file can contain a header line to indicate the name of each field.

title;album;duration;release;artist;type

In this article, we will see how to integrate such a file in Apache Kafka. Of course, we are not going to reinvent the wheel. Many tools already exist to do this and are available in open-source.

We will use the Kafka Connect framework, which is part of the Apache Kafka project. Kafka Connect has been designed to move data in and out of Kafka using connectors.

Kafka Connect File Pulse connector

The Kafka Connect File Pulse connector makes it easy to parse, transform, and stream data file into Kafka. It supports several formats of files, but we will focus on CSV.

For a broad overview of FilePulse, I suggest you read this article :

Kafka Connect FilePulse - One Connector to Ingest them All!

For more information, you can check-out the documentation here.

How to use the connector

The easiest and fastest way to get started with Connect the FilePulse connector is to use the Docker image available on Docker Hub.

$ docker pull streamthoughts/kafka-connect-file-pulse:1.6.3

You can download the docker-compose.yml file available on the GitHub project repository to quickly start a Confluent Platform with Kafka Connect and the FilePulse connector pre-installed.

$ wget https://raw.githubusercontent.com/streamthoughts/kafka-connect-file-pulse/v1.6.3/docker-compose.yml
$ docker-compose up -d

Once all Docker containers are started, you can check that the connector is installed on the Kafka Connect worker accessible on http://localhost:8083.

$ curl -s localhost:8083/connector-plugins|jq '.[].class'|egrep FilePulse

"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector"

You can also install the connector either from GitHub Releases Page or from Confluent Hub.

Ingesting data

Let's create a first connector instance :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-csv-filepulse-00/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.csv$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
        "offset.strategy":"name",
        "skip.headers": "1",
        "topic":"musics-filepulse-csv-00",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1
    }'

You can check the connector is properly started by executing:

$ curl -s localhost:8083/connectors/source-csv-filepulse-00/status|jq  '.connector.state'

"RUNNING"

Right now our connector is not processing any data at all. Let's copy a CSV file into the input directory /tmp/kafka-connect/examples/ :

// Download CSV file
$ export GITHUB_REPO_MASTER=https://raw.githubusercontent.com/streamthoughts/kafka-connect-file-pulse/master/
$ curl -sSL $GITHUB_REPO_MASTER/examples/quickstart-musics-dataset.csv -o musics-dataset.csv

// Copy CSV file to docker-container
$ docker exec -it connect mkdir -p /tmp/kafka-connect/examples
$ docker cp musics-dataset.csv connect://tmp/kafka-connect/examples/musics-dataset-00.csv

Let's check if we have some data into the output topic musics-filepulse-csv-00 :

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t musics-filepulse-csv-00 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

{
  "message": {
    "string": "Zoo Station;Achtung Baby;04:36;1991;U2;Rock"
  },
  "headers": {
    "array": [
      {
        "string": "title;album;duration;release;artist;type"
      }
    ]
  }
}

Note: In the example above, we are using kafkacat to consume messages. The option -o-1 is used to only consume the latest message.

As we can see, our topic contains one Avro message for each line of the CSV file. The reason for this is because we have configured our connector to use the RowFileInputReader (see: task.reader.class configuration)

Each record contains two fields:

message: This field is of type string and represents a single line of the input file.
headers: This field is of type array and contains the first header line of the input file. This field is automatically added by the RowFileInputReader because we have configured it with skip.headers=1.

Headers

The FilePulse connector will add headers to each record containing metadata about the source file from which the data was extracted. This can be useful for debugging but also for data lineage.

$ docker run --tty  --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t musics-filepulse-csv-00 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .headers

[
  "connect.file.name",
  "musics-dataset-00.csv",
  "connect.file.path",
  "/tmp/kafka-connect/examples",
  "connect.file.hash",
  "1466679696",
  "connect.file.size",
  "6588",
  "connect.file.lastModified",
  "1597337097000",
  "connect.hostname",
  "57a2fb6213f9"
]

Parsing data

The File Pulse connector allows us to define complex pipelines to parse, transform, and enrich data through the use of processing Filters.

You can use the built-in DelimitedRowFilter to parse each line. Also, because the first line of the CSV file is a header you can set the property extractColumnName to name the record's fields based on the headers field.

Let's create a new connector with this new configuration :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-csv-filepulse-01/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.csv$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
        "offset.strategy":"name",
        "skip.headers": "1",
        "topic":"musics-filepulse-csv-01",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1,
        "filters":"ParseLine",
        "filters.ParseLine.extractColumnName": "headers",
        "filters.ParseLine.trimColumn": "true",
        "filters.ParseLine.separator": ";",
        "filters.ParseLine.type": "io.streamthoughts.kafka.connect.filepulse.filter.DelimitedRowFilter"
    }'

Then, let's consume the topic musics-filepulse-csv-01 :

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t musics-filepulse-csv-01 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault

{
  "title": {
    "string": "Zoo Station"
  },
  "album": {
    "string": "Achtung Baby"
  },
  "duration": {
    "string": "04:36"
  },
  "release": {
    "string": "1991"
  },
  "artist": {
    "string": "U2"
  },
  "type": {
    "string": "Rock"
  }
}

Filtering data

Sometimes you may want to keep only the lines in a file that have a field with a particular value. For this article, let's imagine we need to only keep the song from the band AC/DC.

For doing this we will extend our filter chain to use the DropFilter and create a new connector (so that our file can be reprocessed):

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-csv-filepulse-02/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.csv$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
        "offset.strategy":"name",
        "skip.headers": "1",
        "topic":"musics-filepulse-csv-02",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1,
        "filters":"ParseLine,KeepACDC",
        "filters.ParseLine.type": "io.streamthoughts.kafka.connect.filepulse.filter.DelimitedRowFilter",
        "filters.ParseLine.extractColumnName": "headers",
        "filters.ParseLine.trimColumn": "true",
        "filters.ParseLine.separator": ";",
        "filters.KeepACDC.type":"io.streamthoughts.kafka.connect.filepulse.filter.DropFilter",
        "filters.KeepACDC.if":"{{ equals($value.artist, '\''AC/DC'\'') }}",
        "filters.KeepACDC.invert":"true"
    }'

The propery if is set with a Simple Connect Expression Language (SCEL) which is a basic expression language provided by the Connect FilePulse connector to access and manipulate records fields.

Finally, you can check that you get the expected result by executing:

$ docker run --tty  --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t musics-filepulse-csv-02 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault.artist

{
  "string": "AC/DC"
}

Changing the field types

You have probably noticed that at no time have we defined the type of our fields. By default, the connector assumes that all fields are of type string and maybe you are happy with that. But in most cases, you will want to convert a few fields. For example, we can convert the field year to be of type "Integer".

For doing this, you can use the AppendFilter with the Simple connect Expression.

Lets' create our final configuration :

$ curl \
    -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-csv-filepulse-03/config \
    -d '{
        "connector.class":"io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
        "fs.scan.directory.path":"/tmp/kafka-connect/examples/",
        "fs.scan.interval.ms":"10000",
        "fs.scan.filters":"io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter",
        "file.filter.regex.pattern":".*\\.csv$",
        "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
        "offset.strategy":"name",
        "skip.headers": "1",
        "topic":"musics-filepulse-csv-03",
        "internal.kafka.reporter.bootstrap.servers": "broker:29092",
        "internal.kafka.reporter.topic":"connect-file-pulse-status",
        "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
        "tasks.max": 1,
        "filters":"ParseLine,KeepACDC,ReleaseToInt",
        "filters.ParseLine.type": "io.streamthoughts.kafka.connect.filepulse.filter.DelimitedRowFilter",
        "filters.ParseLine.extractColumnName": "headers",
        "filters.ParseLine.trimColumn": "true",
        "filters.ParseLine.separator": ";",
        "filters.KeepACDC.type":"io.streamthoughts.kafka.connect.filepulse.filter.DropFilter",
        "filters.KeepACDC.if":"{{ equals($value.artist, '\''AC/DC'\'') }}",
        "filters.KeepACDC.invert":"true",
        "filters.ReleaseToInt.type": "io.streamthoughts.kafka.connect.filepulse.filter.AppendFilter",
        "filters.ReleaseToInt.field": "$value.release",
        "filters.ReleaseToInt.value": "{{ converts($value.release, '\''INTEGER'\'') }}",
        "filters.ReleaseToInt.overwrite": "true"
    }'

Finally, you can check converted field :

$ docker run --tty --network=host edenhill/kafkacat:1.6.0 kafkacat \
        -b localhost:9092 \
        -t musics-filepulse-csv-03 \
        -C -J -q -o-1 -s key=s -s value=avro -r http://localhost:8081 | jq .payload.ConnectDefault.release

{
  "release": {
    "int": 1980
  }
}

The AppendFilter is a very handy filter that allows us to quickly modify a record. For example, we could also use it to set the key of each record to be equal to the album name by adding the following configuration:

"filters.SetKey.type": "io.streamthoughts.kafka.connect.filepulse.filter.AppendFilter",
"filters.SetKey.field": "$key",
"filters.SetKey.value": "{{ uppercase($value.album)}}"

Note: For this article, we have used here the filter chain mechanism provided by the Connect File Pulse connector. But it is also possible to use the Kafka Connect Single Message Transformations (SMT) to perform the same task using the org.apache.kafka.connect.transforms.Cast.

Conclusion

We have seen in this article that it is fairly easy to load a CSV file into Kafka without writing a single line of code using Kafka Connect. Also, the Connect File Pulse connector is a powerful solution that allows you to easily manipulate your data before sending it into Apache Kafka. Please do not hesitate to share this article. If you like this project then add a ⭐ to the GitHub repository to support us. Thank you.

Create Kafka Streams applications faster than ever before with Azkarra Streams

Florian Hussonnois — Thu, 09 Jan 2020 16:02:06 +0000

Kafka Streams is a powerful library for writing streaming applications and microservices on top of Apache Kafka in Java and Scala.

When writing a Kafka Streams application, developers must not only define their topology, i.e. the sequence of operations to be applied to the consumed messages, but also the code needed to execute it.

Furthermore, to write a production-ready application, you will have to know; how to handle processing failures and bad records, how to monitor and operate instances. And, if you plan to expose some internal states, using the Kafka Streams built-in feature so-called "Interactive Queries", you will also have to write code to get access to your data (for example via REST APIs).

As a result of this, your application can quickly become complex with boilerplate code that has no direct business value but that you will have to maintain and duplicate and other projects.

Azkarra Streams is an open-source lightweight Java framework which makes easy to develop and operate Kafka Streams applications (Azkarra is Basque word for "Fast").

Key Features

Azkarra Streams provides a set of features to quickly debug and build production-ready Kafka Streams applications. This includes, among other things:

Lifecycle management of Kafka Streams instances (no more KafkaStreams#start()).
Easy externalization of Topology and KafkaStreams configurations (using Typesafe Config).
Embedded HTTP server for querying state store (Undertow).
HTTP endpoint to monitor streams application metrics (e.g : JSON, Prometheus).
Web UI for topologies visualization.
Encryption and Authentication with SSL or Basic Auth. Etc.

Getting Started

Since Azkarra v0.5.0, one way to get started with Azkarra is to use the official Docker image (streamthoughts/azkarra-streams-worker) that allows running a standalone Azkarra worker to execute one or many Kafka Streams applications.

Azkarra Worker follows the same mechanism used by the Kafka Connect project, i.e. that Kafka Streams topologies are provided as external components that can be started and stopped either via REST calls or an embedded UI.

Let's start an Azkarra worker instance and a broker-single node cluster using the docker-compose.yml available on the GitHub repository.

1 ) Run the following command to download and run containers :

$ curl -s https://raw.githubusercontent.com/streamthoughts/azkarra-streams/master/docker-compose.yml --output \
docker-compose.yml && docker-compose up -d

2 ) Check that Azkarra worker is up and running :

$ curl -sX GET http://localhost:8080 | jq
{
  "azkarraVersion": "0.5.0",
  "commitId": "d2bc2fdc24e68eb143f4388960881974604093ca",
  "branch": "master"
}

3 ) Finally, you can access the Azkarra Web UI is available on: http://localhost:8080/ui.

As we can see, for the moment our worker is doing absolutely nothing since we have not yet deployed a topology. So, let's write a simple Kafka Streams application.

Writing A First Kafka Streams Topology

For demonstrating the use of Azkarra API, we will rewrite the standard WordCountTopology example.

First, let's create a simple Java project and add Azkarra Streams to the dependency of your project.

For Maven (pom.xml):

<dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>2.4.0</version>
        <scope>provided</scope>
    </dependency>

    <dependency>
        <groupId>io.streamthoughts</groupId>
        <artifactId>azkarra-streams</artifactId>
        <version>0.5.0</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

Note that when using Azkarra Worker, your project should never contain any libraries that are provided by Azkarra Worker’s runtime (i.e azkarra-*, kafka-streams).

Secondly, let's define our Kafka Streams Topology by creating a new file WordCountTopology.java.

package azkarra;

import io.streamthoughts.azkarra.api.annotations.*;
import io.streamthoughts.azkarra.api.streams.TopologyProvider;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.*;

import java.util.Arrays;

@Component
@TopologyInfo(description = "WordCount topology example")
public class WordCountTopology implements TopologyProvider {

    @Override
    public Topology get() {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> textLines = builder.stream("streams-plaintext-input");
        textLines
        .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
        .groupBy((key, value) -> value)
        .count(Materialized.as("WordCount"))
        .toStream()
        .to(
            "streams-wordcount-output", 
             Produced.with(Serdes.String(), Serdes.Long())
        );
        return builder.build();
    }

    @Override
    public String version() {
        return "1.0";
    }
}

As you can see, we have implemented the TopologyProvider interface to provide the Topology object. Azkarra enforces you to version each provided topology. This is useful, for example, to execute multiple versions of the same topology or to automatically generate meaningful application.id.

The annotation @Component is required to let Azkarra detects this class.

And that's all! Azkarra will be responsible to create and manage the KafkaStreams instance that will run the provided Topology.

Deploying A Streams Topology

Now, we need to make our WordCountTopology available to the worker.

For doing this, we have to package and install our component into one of the directories configured via the property azkarra.component.paths.

If you look at the docker-compose.yml you will see that this property is set to /tmp/azkarra/components using an environment variable.

The azkarra.component.paths property should define the list of locations (separated by a comma) from which the components will be scanned.

Each configured directories may contain:

an uber JAR containing all of the classes and third-party dependencies for the component (e.g., topology).
a directory containing all JARs for the component

Usually, with Maven, you will use the maven-assembly-plugin or maven-shade-plugin to build your project to an uber JAR.

After packaging your application, you can copy the .jar into the local directory /tmp/azkarra/components.

Then, restart the docker containers as follows:

$ docker-compose restart

Now, you should be able to list the available topologies via the REST API :

curl -sX GET http://localhost:8080/api/v1/topologies | jq 
[
  {
    "name": "azkarra.WordCountTopology",
    "version": "1.0",
    "description": "WordCount topology example",
    "aliases": [
      "WordCount",
      "WordCountTopology"
    ],
    "config": {}
  }
]

Finally, let's start a new Kafka Streams instance by submitting the following JSON config :

curl -H "Content-Type:application/json" \
-X POST http://localhost:8080/api/v1/streams \
--data '{"type": "azkarra.WordCountTopology", "version": "1.0",  "env": "__default", "config": {} }'

In the command above, we are specifying the type and version of the topology to deploy and the target environment.

Indeed, Azkarra has a concept of StreamsExecutionEnvironment which acts as a container for executing streams instances. By default, an environment named __default is created.

Note that Azkarra will automatically create any source and sink topics defined by the topology (azkarra.context.auto.create.topics.enable=true).

Exploring Azkarra Web UI

Azkarra ships with an embedded Web UI that lets you get information about the running Kafka Streams applications.

For example, you can :

Get details about the threads and tasks of a running streams instance :
Visualize the streams topology DAG:
List the Kafka Streams metrics:

Furthermore, the Azkarra Web UI allows you to stop, restart and delete local streams instances.

Querying states stores

Finally, Kafka Streams has a great mechanism to query the states materialized by streams applications via REST API calls.

Let's produce some messages as follows:

$ docker exec -it broker /usr/bin/kafka-console-producer \
--topic streams-plaintext-input \
--broker-list broker:9092

Azkarra Streams
WordCount
I Heart Logs   
Kafka Streams
Making Sense of Stream Processing

The following is an example to query the state WordCount :

curl -sX POST http://localhost:8080/api/v1/applications/word-count-topology-1-0/stores/WordCount \
--data '{"query":{"get":{"key": "streams"}},"type":"key_value", "set_options":{}}' | jq
{
  "took": 1,
  "timeout": false,
  "server": "azkarra:8080",
  "status": "SUCCESS",
  "result": {
    "success": [
      {
        "server": "azkarra:8080",
        "remote": false,
        "records": [
          {
            "key": "streams",
            "value": 2
          }
        ]
      }
    ],
    "total": 1
  }
}

You can also query a state directly through the Azkarra WebUI.

Going further

If you want to read more about using Azkarra Streams, the documentation can be found on GitHub Page.

The documentation contains a step by step getting started to learn basic concepts of Azkarra.

The project also contains some examples.

Conclusion

Azkarra Streams is an initiative to enrich the Kafka Streams ecosystem and facilitate its adoption by developers through a lightweight micro-framework.

We hope this project will be well received by the open-source and Kafka community. Azkarra is still evolving and some features need improvements.
To support Azkarra Streams project, please ⭐ the Github repository or tweet if this project helps you!

Thank you very much!