Kafka static ingest examples
This page provides examples of Kafka static ingest files and flags you can use to test the system
Table of contents
Before you begin
Example 1 - Ingest two values from a Kafka message
Kafka message file
Kafka message with two values
{
"int-kafka-path": 12345,
"string-kafka-path": "arbitraryString"
}
Kafka JSON file
File kafka-static-header-1.json contains the following settings:
[
{
"name": "int-featurebase-name",
"path": [
"int-kafka-path"
],
"type": "int"
},
{
"name": "string-featurebase-name",
"path": [
"string-kafka-path"
],
"type": "string"
}
]
Kafka ingest CLI flags
./molecula-consumer-kafka-static \
--kafka-hosts "localhost:9092" \
--index kafka-test \
--batch-size 10000 \
--topics test-topic \
--max-msgs 10000 \
--auto-generate \
--external-generate \
--header kafka-static-header-1.json
Example 2 - Ingest an array of values from a Kafka message
Kafka message file
Kafka message with array of values:
{
"from_interface": {
"ip": "10.203.33.18",
"port": 38935
},
"to_interface": {
"ip": "203.77.221.220",
"port": 5872
},
"event_time": "2021-06-01T16:02:55Z06:00",
"protocol": "UDP",
"severity": 0,
"bytes": 8593
}
Kafka JSON file
kafka-static-header-2.json reads values from the Kafka message and structures the values according to the destination table.
[
{
"name": "from_ip",
"path": [
"from_interface",
"ip"
],
"type": "string"
},
{
"name": "from_port",
"path": [
"from_interface",
"port"
],
"type": "int"
},
{
"name": "to_ip",
"path": [
"to_interface",
"ip"
],
"type": "string"
},
{
"name": "to_port",
"path": [
"to_interface",
"port"
],
"type": "int"
},
{
"name": "event_time",
"path": [
"event_time"
],
"type": "timestamp"
},
{
"name": "severity",
"path": [
"severity"
],
"type": "set"
},
{
"name": "bytes",
"path": [
"bytes"
],
"type": "int"
},
{
"name": "protocol",
"path": [
"protocol"
],
"type": "string"
}
]
Kafka ingest CLI flags
./molecula-consumer-kafka-static \
--kafka-hosts "localhost:9092" \
--index kafka-test \
--batch-size=10000 \
--topics test-topic \
--auto-generate \
--allow-missing-fields \
--header kafka-static-header-2.json