How do I ingest data from a Kafka static schema?

Ingesting data from a Kafka static schema involves the use of:

• JSON message blob which defines the source data
• JSON document that defines the source data as an array that matches the destination table structure
• Specifying the JSON document using the static CLI flag when running ./molecula-consumer-kafka-static

Before you begin

• Install Apache Kafka on your system
• Install Apache Zookeeper on your system
• Install Apache Avro on your system
• Learn about Kafka static schema settings
• Define your data in a Kafka message/blob

Kafka static schema definition

The schema file is formatted as an array of JSON objects, each of which describes one ingester field.

[
	{
		"name": "<featurebase-field>",
		"path": ["<JSON-config-file>"],
		"type": "<data type>",
		"config": {
			"parameters": "<Optional field data type>"
		}
	}
]

Kafka static JSON parameters

Additional information

• The Schema is specified with a JSON document to ensure compatibility with complex JSON message formats
• The name, path, and type parameters are repeated for each record to import to FeatureBase.

path parameter

• Array indexing is not supported
• Errors are returned if a value is missing from the path
• Missing values can be ignored by adding the --allow-missing-fields flag when using ./molecula-consumer-kafka-static

type data types

This table provides mapping between FeatureBase SQL data types and internal data types used by the application for configuring ingestion, API calls, etc.

Field mapping Avro to IDK

Arrays

Boolean data types

Decimal data type

ID data type

Int data type

Mutex data type

Quantum constraint

Scale data type

String data type

Timestamp Time Date data types

Not supported

Error

UNION

| Properties | Avro | IDK | |—|—|—| | | avro.Union | supports one or two members (if two, one must be avro.NULL) |

Default config parameters

Custom config parameters

The following information applies to "config": "parameters":

Cache

• Improve precision by increasing the cache size for the "ranked" cache type which increases the number of top rows tracked within a shard of data
• Set the data type to "none" to disable the cache.

recordTime fields

recordTime fields have two modes.

TopN cache

TopN cache memory usage is:

• jointly proportional to cache size and number of shards when the cache is full
• for example, the field is greater than the CacheSize rows within each shard)

Disabling TopN cache disables TopN

Querying

Use a TopK or sorted GroupBy when operating a field without a cache.

TIMEQUANTUM

TIMEQUANTUM creates a view of your data based on the specified time. This allows for lower latency queries at the cost of increased storage. For example, set TIMEQUANTUM to:

• MD for queries that include a range of months
• D for queries that include a small number of days

NOTE: Queries run on mismatched time granularities are slower but will function correctly. For example: Querying days on a IDSET or STRINGSET column with TIMEQUANTUM set to YM.

You can omit but not skip time granularity.

• YM is valid
• YH is invalid

TIMEQUANTUM is used when:

• times need to be associated with column data for query purposes
• database space is not at a premium

TTL (Time To Live)

NOTE: FeatureBase recommends using a TTL of 1h or more to improve results.

• TTL enables the deletion of time views where a time range exceeds the stated Time To Live.
• The default TTL of 0s indicates TIMEQUANTUM views will not be deleted.
• TTL runs:
  • when FeatureBase starts and every hour to make view deletion consistent
  • are not guaranteed to run at a specific time

• error: unknown unit is generated if an incorrect value is used (e.g., TTL is set to 60second)

• TTL should not be used if you require complete and consistent historical data.

TTL order of events

This example demonstrates the deletion dates of three column views where TTL is set to 30d

Examples

path example

Select 1 within {"a":{"b":{"c":1}}}

"path": ["a","b","c"]

TopN cache example

This “cache” is used for the TopN approximation. The default setting is:

{
	"CacheType": "ranked",
	"CacheSize": 50000,
}

Ingest two values from a Kafka message

Kafka message with two values

{
    "int-kafka-path": 12345,
    "string-kafka-path": "arbitraryString"
}

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"
    }
]

Execute the following CLI commands from the /featurebase/idk directory.

./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

Ingest data from an array of values

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-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"
    }
]

Execute the following CLI commands from the /featurebase/idk directory.

./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

Further information

• RFC339 definition

Next step

• Kafka static ingest consumer flags reference