> ## Documentation Index
> Fetch the complete documentation index at: https://unstructured-53-docs-243-plugins.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Kafka

<Note>
  If you're new to Unstructured, read this note first.

  Before you can create a source connector, you must first [sign up for Unstructured](https://platform.unstructured.io) and get your
  Unstructured API key. After you sign up, the [Unstructured user interface](/ui/overview) (UI) appears, which you use to get the key.
  To learn how, watch this 40-second [how-to video](https://www.youtube.com/watch?v=FucugLkYB6M).

  After you create the source connector, add it along with a
  [destination connector](/api-reference/workflow/destinations/overview) to a [workflow](/api-reference/workflow/overview#workflows).
  Then run the worklow as a [job](/api-reference/workflow/overview#jobs). To learn how, try out the
  [hands-on Workflow Endpoint quickstart](/api-reference/workflow/overview#quickstart),
  go directly to the [quickstart notebook](https://colab.research.google.com/drive/13f5C9WtUvIPjwJzxyOR3pNJ9K9vnF4ww),
  or watch the two 4-minute video tutorials for the [Unstructured Python SDK](/api-reference/workflow/overview#unstructured-python-sdk).

  You can also create source connectors with the Unstructured user interface (UI).
  [Learn how](/ui/sources/overview).

  If you need help, reach out to the [community](https://short.unstructured.io/pzw05l7) on Slack, or
  [contact us](https://unstructured.io/contact) directly.

  You are now ready to start creating a source connector! Keep reading to learn how.
</Note>

Ingest your files into Unstructured from Kafka.

The requirements are as follows.

* A Kafka cluster in [Confluent Cloud](https://www.confluent.io/confluent-cloud).
  ([Create a cluster](https://docs.confluent.io/cloud/current/clusters/create-cluster.html#create-ak-clusters).)

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

  <iframe width="560" height="315" src="https://www.youtube.com/embed/zcKJ96J4Xvk" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowfullscreen />

* The [hostname and port number](https://docs.confluent.io/cloud/current/clusters/create-cluster.html#view-a-ak-cluster) 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](https://docs.confluent.io/cloud/current/client-apps/topics/index.html#create-topics).
  [Access available topics](https://docs.confluent.io/cloud/current/client-apps/topics/index.html#create-topics).

* For authentication, an [API key and secret](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.html#add-an-api-key).

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

<CodeGroup>
  ```python Python SDK theme={null}
  import os

  from unstructured_client import UnstructuredClient
  from unstructured_client.models.operations import CreateSourceRequest
  from unstructured_client.models.shared import (
      CreateSourceConnector,
      SourceConnectorType,
      KafkaCloudSourceConnectorConfigInput
  )

  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=SourceConnectorType.KAFKA_CLOUD,
                  config=KafkaCloudSourceConnectorConfigInput(
                      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)
  ```

  ```bash curl theme={null}
  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_server": "<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>
      },
  }'
  ```
</CodeGroup>

Replace the preceding placeholders as follows:

* `<name>` (*required*) - A unique name for this connector.
* `<bootstrap-server>` - The hostname of the bootstrap Kafka cluster to connect to.
* `<port>` - The port number of the bootstrap Kafka cluster to connect to. The default is `9092` if not otherwise specified.
* `<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. The default is `default_group_id` if not otherwise specified.
* `<kafka-api-key>` - For authentication, the API key for access to the cluster.
* `<secret>` - For authentication, the secret for access to the cluster.
* `<topic>` - The name of the topic to read messages from or write messages to on the cluster.
* `<batch-size>` (destination connector only) - The maximum number of messages to send in a single batch. The default is `100` if not otherwise specified.
* `<num-messages-to-consume>` (source connector only) - The maximum number of messages that the consumer will try to consume. The default is `100` if not otherwise specified.