Kafka Sink Connector

Note

Despite having a similar name, the Kafka Sink connector is not a GraphDB connector.

Modern businesses have a growing need to integrate data from multiple diverse systems. Automating the update and continuous build of knowledge graphs with the incoming streams of data can be cumbersome due to a number of reasons: verbose functional code writing, numerous transactions per update, suboptimal usability of GraphDB’s RDF mapping language, and the lack of a direct way to stream updates to knowledge graphs.

GraphDB’s open-source Kafka Sink connector, which supports smart updates with SPARQL templates, solves this issue by reducing the amount of code needed for raw event data transformation, making it easier to automate knowledge graph updates. It is a separately running process, which helps to avoid database sizing. The connector allows for customization according to the user’s specific business logic and requires no GraphDB downtime during configuration.

With it, users can push update messages to Kafka, after which a Kafka consumer processes them and applies the updates in GraphDB.

Setup

Important

Before setting up the connector, make sure to have JDK 11 installed.

  1. Install Kafka 3.8.0 or newer.

  2. The Kafka Sink connector can be deployed in a Docker container. To install it and verify that it is working correctly, follow the GitHub README instructions.

Update types

The Kafka Sink connector supports three types of updates:

A given Kafka topic is configured to accept updates in a predefined mode and format. The format must be one of the supported RDF formats.

Simple add

This is a simple INSERT operation where no document identifiers are needed and new data is always added as is. All you need to provide is the new RDF data to add. The following is valid:

  • The Kafka topic is configured to only add data.

  • The Kafka key is irrelevant but it is recommended to use a unique ID such as a random UUID.

  • The Kafka value is the new RDF data to add.

Let’s look at an example scenario.

  1. Start GraphDB on the same or a different machine.

  2. In GraphDB, create a repository called “kafka-test”.

  3. To deploy the connector, execute in the project’s docker-compose directory:

    sudo docker-compose up --scale graphdb=0

    where graphdb=0 denotes that GraphDB must be started outside of the Docker container.

  4. Next, configure the Kafka sink connector that will add data into the repository. In the Kafka sink connector’s directory, execute this curl command:

    curl http://localhost:8083/connectors \
    -H 'Content-Type: application/json' \
    --data '{"name":"kafka-sink-graphdb-add",
        "config":{
         "graphdb.server.url":"http://graphdb.example.com:7200",
         "connector.class":"com.ontotext.kafka.GraphDBSinkConnector",
         "key.converter":"com.ontotext.kafka.convert.DirectRDFConverter",
         "value.converter":"com.ontotext.kafka.convert.DirectRDFConverter",
         "value.converter.schemas.enable":"false",
         "topics":"gdb-add",
         "tasks.max":1,
         "offset.storage.file.filename":"/tmp/storage-add",
         "graphdb.server.repository":"kafka-test",
         "graphdb.batch.size":64,
         "graphdb.batch.commit.limit.ms":1000,
         "graphdb.auth.type":"NONE",
         "graphdb.update.type":"ADD",
         "graphdb.update.rdf.format":"nq"}}'

    This curl command includes the following parameters:

    • topics, which can be one of the following:

      • the name of the topic from which the connector will be reading the documents; in this example, gdb-add

      • a comma-separated list of topics

    • graphdb.server.url: the URL of the GraphDB server; adjust the sample value http://graphdb.example.com:7200 to reflect your server’s location

      Important

      Since GraphDB is running outside the Kafka Sink Docker container, referencing localhost in graphdb.server.url will not work. Use a hostname or IP address that is visible from within the container.

    • graphdb.server.repository: the GraphDB repository in which the connector will write the documents; in the example above, kafka-test

    • graphdb.update.type: the type of the update; ADD above

    This will create the add data connector, which will read documents from the gdb-add topic and send it to the kafka-test repository on the respective GraphDB server.

  5. For the purposes of the example, we will also create a test Kafka producer that will write in the respective Kafka topic. In the Kafka installation directory, execute the following command:

    bin/kafka-console-producer.sh --bootstrap-server localhost:19092 --topic gdb-add

  6. That command will then be waiting for data via standard input, so paste the following RDF data in the same window and press Enter:

    <http://example/subject> <http://example/predicate> "This is an example of adding data" <http://example/graph> .

  7. To finish the data input press Ctrl+Z if you’re using Windows and Ctrl/Cmd+D for any other operating system.

  8. In the Workbench SPARQL editor of the “kafka-test” repository, run the query:

    SELECT * WHERE {
    GRAPH ?g {
        ?s ?p ?o
    }
}

  9. The RDF data that we just added via the producer should be returned as result.

Replace graph

In this update type, a document (the smallest update unit) is defined as the contents of a named graph. To perform an update, the following information must be provided:

  • The IRI of the named graph — the document ID

  • The new RDF contents of the named graph — the document contents

The update is performed with these steps:

  • The Kafka topic is configured for replace graph.

  • The Kafka key defines the named graph to update.

  • The Kafka value defines the contents of the named graph.

Let’s look at an example:

  1. After following the steps above, the Docker container with the Kafka sink connector should already be running, and the “kafka-test” repository will have been created.

  2. Now, let’s configure the Kafka sink connector that will replace data in a named graph. In the Kafka sink connector’s directory, execute this curl command:

    curl http://localhost:8084/connectors \
      -H 'Content-Type: application/json'\
     --data '{"name":"kafka-sink-graphdb-replace",
           "config":{
              "graphdb.server.url":"http://graphdb.example.com:7200",
              "connector.class":"com.ontotext.kafka.GraphDBSinkConnector",
              "key.converter":"com.ontotext.kafka.convert.DirectRDFConverter",
              "value.converter":"com.ontotext.kafka.convert.DirectRDFConverter",
              "value.converter.schemas.enable":"false",
              "topics":"gdb-replace",
              "tasks.max":1,
              "offset.storage.file.filename":"/tmp/storage-replace",
              "graphdb.server.repository":"kafka-test",
              "graphdb.batch.size":64,
              "graphdb.batch.commit.limit.ms":1000,
              "graphdb.auth.type":"NONE",
              "graphdb.update.type":"REPLACE_GRAPH",
              "graphdb.update.rdf.format":"nq"}}'

    This uses the same important parameters as in the add data example above.

    This will configure the replace graph connector, which will read data from the gdb-replace topic and send it to the kafka-test repository on the respective GraphDB server.

  3. To replace data in a specific named graph, you need to provide:

    • the name of the graph as the key of the Kafka message in the Kafka topic

    • the new data for the replace operation as the value of the Kafka message

    So, you need to modify the producer to create a key-value message. In the Kafka installation directory, execute the following command:

    bin/kafka-console-producer.sh --bootstrap-server localhost:19092 --topic gdb-replace \
    --property parse.key=true --property key.separator="-"

  4. To replace the data in the graph, paste the following into the same window and then press Enter and either Ctrl+Z or Ctrl/Cmd+D like before to show that you are done entering data:

    http://example/graph-<http://example/subject> <http://example/predicate> "Successfully replaced graph" <http://example/graph> .

    The key value must be the ?id value from the template.

  5. To see the replaced data, run the query from above in the Workbench SPARQL editor of the “kafka-test” repository:

    SELECT * WHERE {
    GRAPH ?g {
        ?s ?p ?o
    }
}

  6. The replaced data in the graph should be returned as result.

DELETE/INSERT template

In this update type, a document is defined as all triples for a given document identifier according to a predefined schema. The schema is described as a SPARQL DELETE/INSERT template that can be filled from the provided data at update time. The following must be present at update time:

  • The SPARQL template update, which must be predefined, not provided at update time

    • this can be a DELETE WHERE update that only deletes the previous version of the document and the new data is inserted as is

    • this can be a DELETE INSERT WHERE update that deletes the previous version of the document and adds additional triples such as timestamp information

  • The IRI of the updated document

  • The new RDF contents of the updated document

The update is performed with these steps:

  • The Kafka topic is configured for a specific template.

  • The Kafka value defines the data to be updated for the bound ID.

Important

One instance configuration can work with one template. If you need to use a second template, you can create a second configuration on the same instance.

Let’s look at an example scenario:

  1. As with the previous update, the Docker container with the Kafka Sink connector running is already created, and the “kafka-test” repository is used for the purposes of the example.

  2. With this type of update, you need to first create a SPARQL template for the data update.

    1. In Workbench, go to Setup ‣ SPARQL Templates ‣ Create new SPARQL template.

    2. Enter a template IRI (required) such as http://example.com/my-template.

    3. Go to Import ‣ User data ‣ Import RDF text snippet and import the following sample data describing five fictitious wines:

      @prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
@prefix wine: <http://www.example.com/demo/wine#> .

wine:RedWine rdfs:subClassOf wine:Wine .
wine:WhiteWine rdfs:subClassOf wine:Wine .
wine:RoseWine rdfs:subClassOf wine:Wine .

wine:Merlot
    rdf:type wine:Grape ;
    rdfs:label "Merlot" .

wine:CabernetSauvignon
    rdf:type wine:Grape ;
    rdfs:label "Cabernet Sauvignon" .

wine:CabernetFranc
    rdf:type wine:Grape ;
    rdfs:label "Cabernet Franc" .

wine:PinotNoir
    rdf:type wine:Grape ;
    rdfs:label "Pinot Noir" .

wine:Chardonnay
    rdf:type wine:Grape ;
    rdfs:label "Chardonnay" .

wine:Yoyowine
    rdf:type wine:RedWine ;
    wine:madeFromGrape wine:CabernetSauvignon ;
    wine:hasSugar "dry" ;
    wine:hasYear "2013"^^xsd:integer .

wine:Franvino
    rdf:type wine:RedWine ;
    wine:madeFromGrape wine:Merlot ;
    wine:madeFromGrape wine:CabernetFranc ;
    wine:hasSugar "dry" ;
    wine:hasYear "2012"^^xsd:integer .

wine:Noirette
    rdf:type wine:RedWine ;
    wine:madeFromGrape wine:PinotNoir ;
    wine:hasSugar "medium" ;
    wine:hasYear "2012"^^xsd:integer .

wine:Blanquito
    rdf:type wine:WhiteWine ;
    wine:madeFromGrape wine:Chardonnay ;
    wine:hasSugar "dry" ;
    wine:hasYear "2012"^^xsd:integer .

wine:Rozova
    rdf:type wine:RoseWine ;
    wine:madeFromGrape wine:PinotNoir ;
    wine:hasSugar "medium" ;
    wine:hasYear "2013"^^xsd:integer .

    4. Next, navigate to Setup ‣ SPARQL Templates ‣ Create new SPARQL template to create the SPARQL template:

      PREFIX wine: <http://www.ontotext.com/example/wine#>
PREFIX xsd:  <http://www.w3.org/2001/XMLSchema#>

DELETE {
    ?s wine:hasSugar ?oldValue .
    ?s wine:hasYear ?oldYear
} INSERT {
    ?s wine:hasSugar ?sugar .
    ?s wine:hasYear ?year .
} WHERE {
    ?s ?p ?oldValue .
    ?s ?p1 ?oldYear .
}

    5. Save it, after which it will appear in the templates list.

      _images/kafka-sink-update-template.png

Now you need to configure the Kafka sink connector that will update some data in a named graph. In the directory of the Kafka sink connector, execute:

curl http://localhost:8085/connectors \
    -H 'Content-Type: application/json' \
    --data '{"name": "kafka-sink-graphdb-update",
        "config": {
        "graphdb.server.url":"http://graphdb.example.com:7200",
        "connector.class": "com.ontotext.kafka.GraphDBSinkConnector",
        "key.converter": "com.ontotext.kafka.convert.DirectRDFConverter",
        "value.converter": "com.ontotext.kafka.convert.DirectRDFConverter",
        "value.converter.schemas.enable": "false",
        "topics": "gdb-update",
        "tasks.max": 1,
        "offset.storage.file.filename": "/tmp/storage-update",
        "graphdb.server.repository": "kafka-test",
        "graphdb.batch.size": 64,
        "graphdb.batch.commit.limit.ms": 1000,
        "graphdb.auth.type": "NONE",
        "graphdb.update.type": "SMART_UPDATE",
        "graphdb.update.rdf.format": "jsonld",
        "graphdb.template.id":
        "http://example.com/my-template"}}'

Note how the data passed by the curl command specifies the template IRI "graphdb.template.id":"http://example.com/my-template" that was created in Workbench earlier, which the sink connector will execute.

  1. In the Kafka installation directory, execute the following command:

    bin/kafka-console-producer.sh --bootstrap-server localhost:19092 --topic gdb-update

  2. To execute an update, paste the following into the same window and then press Enter and either Ctrl+Z or Ctrl/Cmd+D like before to show that you are done entering data:

    {"sugar" : "none" , "year" : 2020 , "s" : "http://www.ontotext.com/example/wine#Blanquito" }

  3. To see the updated data, run the query from above in the Workbench SPARQL editor of the “kafka-test” repository:

    PREFIX wine: <http://www.ontotext.com/example/wine#>

SELECT ?s ?p ?o
WHERE {
   BIND(wine:Blanquito as ?s ) .
    ?s ?p ?o .
}

  4. The updated data in the graph should be returned as result.

Configuration properties

The following properties are used to configure the Kafka Sink connector:

Property

Description

Valid values

name

Globally unique name to use for this connector.

String

connector.class

Name or alias of the class for this connector. Must be a subclass of org.apache.kafka.connect.connector.Connector. If the connector is org.apache.kafka.connect.file.FileStreamSinkConnector, you can either specify this full name or use FileStreamSink or FileStreamSinkConnector to make the configuration a bit shorter.

String

key.converter

Converter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the keys in messages written to or read from Kafka, and since this is independent of connectors, it allows any connector to work with any serialization format. Default is NULL.

Class

value.converter

Converter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the values (as opposed to keys) in messages written to or read from Kafka, and since this is independent of connectors, it allows any connector to work with any serialization format. Default is NULL.

Class

topics

Comma-separated list of topics to consume.

List

tasks.max

Maximum number of tasks to use for this connector. Default is 1.

Integer

graphdb.server.url

The URL of the GraphDB server.

String

graphdb.server.repository

The GraphDB repository where the connector will write the documents.

String

graphdb.batch.size

The maximum number of documents to send from Kafka to GraphDB in one transaction.

Integer

graphdb.auth.type

The authentication type.

NONE, BASIC, CUSTOM

graphdb.auth.basic.username

The username for basic authentication.

String

graphdb.auth.basic.password

The password for basic authentication.

String

graphdb.auth.header.token

The GraphDB authentication token.

String

graphdb.update.type

The type of the transaction.

ADD, REPLACE_GRAPH, SMART_UPDATE

graphdb.update.rdf.format

The format of the documents sent from Kafka to GraphDB. Default is ttl.

Any supported RDF format file extension. For example, to send Turtle-star to GraphDB, set this to ttls, or to send JSON-LD, set it to jsonld.

graphdb.batch.commit.limit.ms

The timeout applied per batch that is not full before it is committed. Default is 3000.

Long

graphdb.tls.thumbprint

The SHA-256 thumbprint of the GraphDB certificate that the connector should import and trust.

String

graphdb.tls.hostname.verification.enabled

Enables the hostname verification against the certificate or certificate chain. Default is true.

Boolean

errors.tolerance

Behavior for tolerating errors during connector operation. NONE is the default value and signals that any error will result in an immediate connector task failure; ALL changes the behavior to skip over problematic records.

NONE, ALL

errors.deadletterqueue.topic.name

The topic name in Kafka brokers to store failed records. Default is blank.

String

errors.deadletterqueue.topic.replication.factor

Replication factor used to create the dead letter queue topic when it does not already exist. Default is 3.

Short

errors.retry.timeout

The maximum duration in milliseconds that a failed operation will be reattempted. Default is 0, which means no retries will be attempted. Use -1 for infinite retries.

Long

errors.retry.delay.max.ms

The maximum duration in milliseconds between consecutive retry attempts.

Long

bootstrap.servers

The IRIs of Kafka brokers. Defaults are localhost:9092, localhost:9093, localhost:9093.

List