Kafka delete message reference

These instructions apply to Kafka schemas managed by Confluent Schema Management.

FeatureBase recommends using Confluent Schema Management because it makes it easier to setup Kafka dependencies in a local environment:

• Schema registry
• Apache Kafka
• Apache Zookeeper

This reference page provides information on structuring an Avro encoded and formatted Kafka messages that is:

• referenced by the Kafka delete consumer CLI command
• used to delete values from a FeatureBase database

Before you begin

• Learn about the Confluent Schema Registry
• Setup Python in a virtual environment
• Install the Confluent Kafka Python Client
• Learn how to setup Confluent and Kafka topics to store your data
• Create at least one Kafka topic
• Encode Kafka records using Apache Avro
• Confluent Schema Registry running somewhere the Kafka consumer can connect to it over HTTP or HTTPS

Kafka delete message syntax

{
    "namespace": "<example.test>",
    "type": "<delete-property>",
    "name": "<schema_name>",
    "delete": "<delete-property>",
    "fields":
      [
        <Kafka-Avro-data-types>,
      ]
}

Kafka message delete properties

Additional information

Delete fields properties

./molecula-consumer-kafka-delete defaults to "delete": "fields" if the "delete" property is not defined in the JSON message.

Set "delete": "fields" to delete all values:

• in a list of fields
• for a single record
• for IDSET or STRINGSET fields where you don’t know the specific values to delete

"delete": "fields" requires:

• FeatureBase index configured with keys: true
• {"type": "array", "items": "string"} defined under the "fields" parameter in the Kafka message

Delete values properties

Set "delete": "values" to:

• delete specified values from IDSET and STRINGSET fields
• specify null value for BOOL, DECIMAL, INT and TIMESTAMP fields

"delete": "values" requires:

• Avro fields containing data that can be deleted
• Matching name for Avro Record Schema field and target FeatureBase record

Avro data type field mapping

Kafka delete is not supported for the FeatureBase time or time-quantum data types.

Avro data types to Kafka message

| Avro data type | Kafka message | Additional | |—|—|—| | string
int | Values to delete from FeatureBase field | Retain field values by using the union data type to union null value with string or int data types | | boolean | true when FeatureBase field value should be deleted |

Delete records parameters

The "delete": "records" parameter requires one or more Avro field data types:

Ignore missing fields

All Kafka Avro fields are required.

However, you can ignore missing fields and leave existing values untouched by:

• creating a union with null data type in the Kafka Avro JSON config file
• adding the allow-missing-fields flag to the ./molecula-consumer-kafka CLI command

Kafka Avro data type syntax

Map Avro field data types and property key-value pairs to determine the FeatureBase record data type

BOOL

DECIMAL

ID

IDSET

IDSETQ

STRINGSETQ Avro strings require a matching RecordTime field in the Avro Schema. Examples are provided below.

INT

Strings

STRINGSET

STRINGSETQ

STRINGSETQ Avro strings require a matching RecordTime field in the Avro Schema. Examples are provided below.

TIMESTAMP

Examples

Kafka delete fields

Kafka delete field Avro schema definition

{
    "namespace": "example.test",
    "type": "record",
    "name": "a_delete_schema",
    "delete": "fields",
    "fields": [
        {"name": "pk0", "type": "string"},
        {"name": "pk1", "type": "string"},
        {"name": "fields", "type": {"type": "array", "items": "string"}}
    ]
}

Kafka delete field raw message

This message:

• sets int_fld to null
• clears all the values that were in stringset_fld.

{"pk0": "9z4aw", "pk1": "5ptDx", "fields": ["int_fld","stringset_fld"]}

Kafka delete field ingester configuration

./molecula-consumer-kafka-delete \
    --primary-key-fields "pk0,pk1" \
    --topics delete_topic \
    --kafka-bootstrap-server localhost:9092 \
    --schema-registry-url localhost:8081 \
    --featurebase-hosts localhost:10101 \
    --featurebase-grpc-hosts localhost:20101 \
    --index an_index

Kafka delete values

Kafka delete values Avro schema definition

{
    "namespace": "example.test",
    "type": "record",
    "name": "delete_value_schema",
    "doc": "All supported avro types and property variations",
    "delete": "values",
    "fields": [
        {"name": "_id", "type": "string"},
        {"name": "stringset_string", "type": ["string", "null"]},
        {"name": "string_string", "type": ["string", "null"]},
        {"name": "stringset_stringarray", "type": [{"type": "array", "items": "string"}, "null"]},
        {"name": "idset_int", "type": ["int", "null"]},
        {"name": "id_int", "type": ["int", "null"]},
        {"name": "idset_intarray", "type": [{"type": "array", "items": "int"}, "null"]},
        {"name": "int_int", "type": "boolean"},
        {"name": "decimal_double", "type": "boolean"},
        {"name": "bools", "type": [{"type": "array", "items": "string"}, "null"]},
        {"name": "timestamp_bytes_int", "type": "boolean"}
    ]
}

Raw Kafka Messages

{"_id": "u2Yr4|sHaUv|x5z8P", "stringset_string": {"null": null}, "string_string": {"string": "ZgkOB"}, "stringset_stringarray": {"array": ["u2Yr4", "PYE8V", "VBcyJ"]}, "idset_int": {"int": 890}, "id_int": {"int": 39}, "idset_intarray": {"array": [731, 13]}, "int_int": false, "bools": {"array": ["bool_bool"]}, "decimal_double": true, "timestamp_bytes_int": true}
{"_id": "h1iqc|5ptDx|iYeOV", "stringset_string": {"null": null}, "string_string": {"null": null}, "stringset_stringarray": {"null": null}, "idset_int": {"null": null}, "id_int": {"null": null}, "idset_intarray": {"null": null}, "int_int": true,  "decimal_double": true, "bools": {"null": null}, "timestamp_bytes_int": true}

Kafka delete consumer configuration

./molecula-consumer-kafka-delete \
    --topics delete_topic \
    --kafka-bootstrap-server localhost:9092 \
    --schema-registry-url localhost:8081 \
    --featurebase-hosts localhost:10101 \
    --featurebase-grpc-hosts localhost:20101 \
    --index an_index

The Avro _id has a string data type therefore --index an_index will have keys: true

Kafka delete records

Kafka delete records Avro schema

{
    "namespace": "example.test",
    "type": "record",
    "name": "alltypes_delete_records",
    "docs": "supply list of keys or a PQL filter",
    "delete": "records",
    "fields": [
        {"name": "keys", "type": [{"type": "array", "items": "string"}, "null"]},
        {"name": "filter", "type": ["string", "null"]}
    ]
}

Kafka delete records raw message

{"keys": {"array": ["9z4aw|5ptDx|CKs1F", "ASSAw|kauLy|oxjI0"]}, "filter": {"null": null}}
{"keys": {"null": null}, "filter": {"string": "Row(stringset_string='58KIR')"}}

• The first message deletes the entire record for records with ID 9z4aw|5ptDx|CKs1F and ASSAw|kauLy|oxjI0.
• The second message deletes all records that had the value 58KIR in the field stringset_string.

Kafka delete records ingest definition

./molecula-consumer-kafka-delete \
    --topics delete_topic \
    --kafka-bootstrap-server localhost:9092 \
    --schema-registry-url localhost:8081 \
    --featurebase-hosts localhost:10101 \
    --featurebase-grpc-hosts localhost:20101 \
    --index an_index

Next step

• Kafka Avro delete ingest flags