Kafka

First time creating a connector? Read this first.

Ingest your files into Unstructured from Kafka.

Requirements

You will need:

• A Kafka cluster in Confluent Cloud. (Create a cluster.)

  The following video shows how to set up a Kafka cluster in Confluent Cloud:

• The hostname and port number of the bootstrap Kafka cluster to connect to..

• The name of the topic to read messages from or write messages to on the cluster. Create a topic. Access available topics.

• For authentication, an API key and secret.

Examples

To create a Kafka source connector, see the following examples.

For more information on working with source connectors using the Unstructured API, see Source endpoints.

Python SDK

import os

from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateSourceRequest
from unstructured_client.models.shared import CreateSourceConnector

with UnstructuredClient(api_key_auth=os.getenv("UNSTRUCTURED_API_KEY")) as client:
    response = client.sources.create_source(
        request=CreateSourceRequest(
            create_source_connector=CreateSourceConnector(
                name="<name>",
                type="kafka-cloud",
                config={
                    "bootstrap_servers": "<bootstrap-server>",
                    "port": <port>,
                    "group_id": "<group-id>",
                    "kafka_api_key": "<kafka-api-key>",
                    "secret": "<secret>",
                    "topic": "<topic>",
                    "num_message_to_consume": <num-message-to-consume>
                }
            )
        )
    )

    print(response.source_connector_information)

curl --request 'POST' --location \
"$UNSTRUCTURED_API_URL/sources" \
--header 'accept: application/json' \
--header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
--header 'content-type: application/json' \
--data \
'{
    "name": "<name>",
    "type": "kafka-cloud",
    "config": {
        "bootstrap_servers": "<bootstrap-server>",
        "port": <port>,
        "group_id": "<group-id>",
        "kafka_api_key": "<kafka-api-key>",
        "secret": "<secret>",
        "topic": "<topic>",
        "num_messages_to_consume": <num-messages-to-consume>
    },
}'

Configuration settings

Replace the preceding placeholders as follows:

name · string · required

A unique name for the connector.

bootstrap_servers · string

The hostname of the bootstrap Kafka cluster to connect to.

port · integer · default: "9092"

The port number of the bootstrap Kafka cluster to connect to.

group_id · string · default: "default_group_id"

The ID of the consumer group. A consumer group is a way to allow a pool of consumers to divide the consumption of data over topics and partitions.

kafka_api_key · string

For authentication, the API key for access to the cluster.

secret · string

For authentication, the secret for access to the cluster.

topic · string

The name of the topic to read messages from or write messages to on the cluster.

batch_size · integer · default: "100"

Destination connector only. The maximum number of messages to send in a single batch.

num_messages_to_consume · integer · default: "100"

Source connector only. The maximum number of messages that the consumer will try to consume.

Learn more

•   Unstructured Platform Now Integrates with Apache Kafka in Confluent Cloud