# Kafka Integration (`brasil-scrapers/kafka-integration`)

This integration facilitates the processing of data from an Apify dataset and its delivery to a Kafka topic. It is configured via a schema that includes details about the dataset, Kafka configuration, and batch processing size.

- **URL**: https://beta-securitybyobscurity.apify.com/brasil-scrapers/kafka-integration.md
- **Developed by:** [Brasil Scrapers](https://beta-securitybyobscurity.apify.com/brasil-scrapers) (community)
- **Categories:** Integrations, Automation
- **Stats:** 2 total users, 1 monthly users, 100.0% runs succeeded, 1 bookmarks
- **User rating**: No ratings yet

## Pricing

$10.00/month + usage

To use this Actor, you pay a monthly rental fee to the developer. The rent is subtracted from your prepaid usage every month after the free trial period.
You also pay for the Apify platform usage, which gets cheaper the higher Apify subscription plan you have.

Learn more: https://docs.apify.com/platform/actors/running/actors-in-store#rental-actors

# README

## Apify to Kafka Integration

### Overview

This integration facilitates the processing of data from an Apify dataset and its delivery to a Kafka topic. It is configured via a schema that includes details about the dataset, Kafka configuration, and batch processing size.

### Input Schema

The input schema defines the configuration required to run the integration effectively. Below are the details of the schema and its properties:

#### Schema Details

- **Title**: Apify to Kafka Input Schema
- **Type**: Object
- **Schema Version**: 1

#### Properties

##### 1. **Default Dataset ID**

- **Title**: Default Dataset ID
- **Type**: String
- **Description**: The ID of the Apify dataset to process.
- **Default**: `default`
- **Editor**: Textfield

##### 2. **Kafka Configuration**

- **Title**: Kafka Configuration
- **Type**: Object
- **Description**: Configuration settings for Kafka connection.
- **Editor**: JSON
- **Default**:
  ```json
  {
      "clientId": "apify-kafka-producer",
      "brokers": ["localhost:9092"],
      "topic": "test-topic",
      "ssl": false
  }
  ```
- **Properties**:
  - **Client ID**:
    - **Title**: Client ID
    - **Type**: String
    - **Description**: Kafka client identifier.
    - **Default**: `apify-kafka-producer`
  - **Brokers**:
    - **Title**: Brokers
    - **Type**: Array of strings
    - **Description**: Array of Kafka broker addresses.
    - **Default**: `["localhost:9092"]`
  - **Topic**:
    - **Title**: Topic
    - **Type**: String
    - **Description**: Kafka topic name.
    - **Default**: `test-topic`
  - **SSL**:
    - **Title**: SSL
    - **Type**: Boolean
    - **Description**: Enable/disable SSL connection.
    - **Default**: `false`
  - **SASL** (Optional):
    - **Title**: SASL
    - **Type**: Object
    - **Description**: SASL configuration for Kafka connection.
    - **Properties**:
      - **Username**:
        - **Title**: Username
        - **Type**: String
        - **Description**: Kafka SASL username.
      - **Password**:
        - **Title**: Password
        - **Type**: String
        - **Description**: Kafka SASL password.
      - **Mechanism**:
        - **Title**: Mechanism
        - **Type**: String
        - **Description**: Kafka SASL mechanism.
        - **Default**: `plain`

##### 3. **Batch Size**

- **Title**: Batch Size
- **Type**: Integer
- **Description**: Number of messages to process in each batch.
- **Default**: `2`
- **Minimum**: `1`

#### Required Properties

- `defaultDatasetId`
- `kafkaConfig`

### Example Configuration

```json
{
    "defaultDatasetId": "my-dataset-id",
    "kafkaConfig": {
        "clientId": "my-kafka-client",
        "brokers": ["kafka-broker1:9092", "kafka-broker2:9092"],
        "topic": "my-kafka-topic",
        "ssl": true,
        "sasl": {
            "username": "my-username",
            "password": "my-password",
            "mechanism": "plain"
        }
    },
    "batchSize": 5
}
```

### How to Use

1. **Define the Input**: Provide the necessary configurations in JSON format as per the schema.
2. **Run the Integration**: Pass the configuration to the Apify actor or script responsible for processing and delivering the dataset to Kafka.
3. **Monitor Outputs**: Check your Kafka topic for incoming messages based on the processed dataset.

### Notes

- Ensure the Kafka broker addresses are reachable from the environment where the integration is executed.
- SASL configuration is optional but required for secure Kafka connections.
- Modify the batch size according to the volume of data and processing capacity.

# Actor input Schema

## `defaultDatasetId` (type: `string`):

The ID of the Apify dataset to process

## `kafkaConfig` (type: `object`):

Configuration settings for Kafka connection

## `batchSize` (type: `integer`):

Number of messages to process in each batch

## Actor input object example

```json
{
  "defaultDatasetId": "default",
  "kafkaConfig": {
    "clientId": "apify-kafka-producer",
    "brokers": [
      "localhost:9092"
    ],
    "topic": "test-topic",
    "ssl": false
  },
  "batchSize": 2
}
```

# API

You can run this Actor programmatically using our API. Below are code examples in JavaScript, Python, and CLI, as well as the OpenAPI specification and MCP server setup.

## JavaScript example

```javascript
import { ApifyClient } from 'apify-client';

// Initialize the ApifyClient with your Apify API token
// Replace the '<YOUR_API_TOKEN>' with your token
const client = new ApifyClient({
    token: '<YOUR_API_TOKEN>',
});

// Prepare Actor input
const input = {
    "kafkaConfig": {
        "clientId": "apify-kafka-producer",
        "brokers": [
            "localhost:9092"
        ],
        "topic": "test-topic",
        "ssl": false
    }
};

// Run the Actor and wait for it to finish
const run = await client.actor("brasil-scrapers/kafka-integration").call(input);

// Fetch and print Actor results from the run's dataset (if any)
console.log('Results from dataset');
console.log(`💾 Check your data here: https://console.apify.com/storage/datasets/${run.defaultDatasetId}`);
const { items } = await client.dataset(run.defaultDatasetId).listItems();
items.forEach((item) => {
    console.dir(item);
});

// 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/js/docs

```

## Python example

```python
from apify_client import ApifyClient

# Initialize the ApifyClient with your Apify API token
# Replace '<YOUR_API_TOKEN>' with your token.
client = ApifyClient("<YOUR_API_TOKEN>")

# Prepare the Actor input
run_input = { "kafkaConfig": {
        "clientId": "apify-kafka-producer",
        "brokers": ["localhost:9092"],
        "topic": "test-topic",
        "ssl": False,
    } }

# Run the Actor and wait for it to finish
run = client.actor("brasil-scrapers/kafka-integration").call(run_input=run_input)

# Fetch and print Actor results from the run's dataset (if there are any)
print("💾 Check your data here: https://console.apify.com/storage/datasets/" + run["defaultDatasetId"])
for item in client.dataset(run["defaultDatasetId"]).iterate_items():
    print(item)

# 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/python/docs/quick-start

```

## CLI example

```bash
echo '{
  "kafkaConfig": {
    "clientId": "apify-kafka-producer",
    "brokers": [
      "localhost:9092"
    ],
    "topic": "test-topic",
    "ssl": false
  }
}' |
apify call brasil-scrapers/kafka-integration --silent --output-dataset

```

## MCP server setup

```json
{
    "mcpServers": {
        "apify": {
            "command": "npx",
            "args": [
                "mcp-remote",
                "https://mcp.apify.com/?tools=brasil-scrapers/kafka-integration",
                "--header",
                "Authorization: Bearer <YOUR_API_TOKEN>"
            ]
        }
    }
}

```

## OpenAPI specification

```json
{
    "openapi": "3.0.1",
    "info": {
        "title": "Kafka Integration",
        "description": "This integration facilitates the processing of data from an Apify dataset and its delivery to a Kafka topic. It is configured via a schema that includes details about the dataset, Kafka configuration, and batch processing size.",
        "version": "0.1",
        "x-build-id": "7wMvSz4nuM84KFlHd"
    },
    "servers": [
        {
            "url": "https://api.apify.com/v2"
        }
    ],
    "paths": {
        "/acts/brasil-scrapers~kafka-integration/run-sync-get-dataset-items": {
            "post": {
                "operationId": "run-sync-get-dataset-items-brasil-scrapers-kafka-integration",
                "x-openai-isConsequential": false,
                "summary": "Executes an Actor, waits for its completion, and returns Actor's dataset items in response.",
                "tags": [
                    "Run Actor"
                ],
                "requestBody": {
                    "required": true,
                    "content": {
                        "application/json": {
                            "schema": {
                                "$ref": "#/components/schemas/inputSchema"
                            }
                        }
                    }
                },
                "parameters": [
                    {
                        "name": "token",
                        "in": "query",
                        "required": true,
                        "schema": {
                            "type": "string"
                        },
                        "description": "Enter your Apify token here"
                    }
                ],
                "responses": {
                    "200": {
                        "description": "OK"
                    }
                }
            }
        },
        "/acts/brasil-scrapers~kafka-integration/runs": {
            "post": {
                "operationId": "runs-sync-brasil-scrapers-kafka-integration",
                "x-openai-isConsequential": false,
                "summary": "Executes an Actor and returns information about the initiated run in response.",
                "tags": [
                    "Run Actor"
                ],
                "requestBody": {
                    "required": true,
                    "content": {
                        "application/json": {
                            "schema": {
                                "$ref": "#/components/schemas/inputSchema"
                            }
                        }
                    }
                },
                "parameters": [
                    {
                        "name": "token",
                        "in": "query",
                        "required": true,
                        "schema": {
                            "type": "string"
                        },
                        "description": "Enter your Apify token here"
                    }
                ],
                "responses": {
                    "200": {
                        "description": "OK",
                        "content": {
                            "application/json": {
                                "schema": {
                                    "$ref": "#/components/schemas/runsResponseSchema"
                                }
                            }
                        }
                    }
                }
            }
        },
        "/acts/brasil-scrapers~kafka-integration/run-sync": {
            "post": {
                "operationId": "run-sync-brasil-scrapers-kafka-integration",
                "x-openai-isConsequential": false,
                "summary": "Executes an Actor, waits for completion, and returns the OUTPUT from Key-value store in response.",
                "tags": [
                    "Run Actor"
                ],
                "requestBody": {
                    "required": true,
                    "content": {
                        "application/json": {
                            "schema": {
                                "$ref": "#/components/schemas/inputSchema"
                            }
                        }
                    }
                },
                "parameters": [
                    {
                        "name": "token",
                        "in": "query",
                        "required": true,
                        "schema": {
                            "type": "string"
                        },
                        "description": "Enter your Apify token here"
                    }
                ],
                "responses": {
                    "200": {
                        "description": "OK"
                    }
                }
            }
        }
    },
    "components": {
        "schemas": {
            "inputSchema": {
                "type": "object",
                "required": [
                    "defaultDatasetId",
                    "kafkaConfig"
                ],
                "properties": {
                    "defaultDatasetId": {
                        "title": "Default Dataset ID",
                        "type": "string",
                        "description": "The ID of the Apify dataset to process",
                        "default": "default"
                    },
                    "kafkaConfig": {
                        "title": "Kafka Configuration",
                        "type": "object",
                        "description": "Configuration settings for Kafka connection",
                        "default": {
                            "clientId": "apify-kafka-producer",
                            "brokers": [
                                "localhost:9092"
                            ],
                            "topic": "test-topic",
                            "ssl": false
                        }
                    },
                    "batchSize": {
                        "title": "Batch Size",
                        "minimum": 1,
                        "type": "integer",
                        "description": "Number of messages to process in each batch",
                        "default": 2
                    }
                }
            },
            "runsResponseSchema": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "object",
                        "properties": {
                            "id": {
                                "type": "string"
                            },
                            "actId": {
                                "type": "string"
                            },
                            "userId": {
                                "type": "string"
                            },
                            "startedAt": {
                                "type": "string",
                                "format": "date-time",
                                "example": "2025-01-08T00:00:00.000Z"
                            },
                            "finishedAt": {
                                "type": "string",
                                "format": "date-time",
                                "example": "2025-01-08T00:00:00.000Z"
                            },
                            "status": {
                                "type": "string",
                                "example": "READY"
                            },
                            "meta": {
                                "type": "object",
                                "properties": {
                                    "origin": {
                                        "type": "string",
                                        "example": "API"
                                    },
                                    "userAgent": {
                                        "type": "string"
                                    }
                                }
                            },
                            "stats": {
                                "type": "object",
                                "properties": {
                                    "inputBodyLen": {
                                        "type": "integer",
                                        "example": 2000
                                    },
                                    "rebootCount": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "restartCount": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "resurrectCount": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "computeUnits": {
                                        "type": "integer",
                                        "example": 0
                                    }
                                }
                            },
                            "options": {
                                "type": "object",
                                "properties": {
                                    "build": {
                                        "type": "string",
                                        "example": "latest"
                                    },
                                    "timeoutSecs": {
                                        "type": "integer",
                                        "example": 300
                                    },
                                    "memoryMbytes": {
                                        "type": "integer",
                                        "example": 1024
                                    },
                                    "diskMbytes": {
                                        "type": "integer",
                                        "example": 2048
                                    }
                                }
                            },
                            "buildId": {
                                "type": "string"
                            },
                            "defaultKeyValueStoreId": {
                                "type": "string"
                            },
                            "defaultDatasetId": {
                                "type": "string"
                            },
                            "defaultRequestQueueId": {
                                "type": "string"
                            },
                            "buildNumber": {
                                "type": "string",
                                "example": "1.0.0"
                            },
                            "containerUrl": {
                                "type": "string"
                            },
                            "usage": {
                                "type": "object",
                                "properties": {
                                    "ACTOR_COMPUTE_UNITS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "DATASET_READS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "DATASET_WRITES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "KEY_VALUE_STORE_READS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "KEY_VALUE_STORE_WRITES": {
                                        "type": "integer",
                                        "example": 1
                                    },
                                    "KEY_VALUE_STORE_LISTS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "REQUEST_QUEUE_READS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "REQUEST_QUEUE_WRITES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "DATA_TRANSFER_INTERNAL_GBYTES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "DATA_TRANSFER_EXTERNAL_GBYTES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "PROXY_RESIDENTIAL_TRANSFER_GBYTES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "PROXY_SERPS": {
                                        "type": "integer",
                                        "example": 0
                                    }
                                }
                            },
                            "usageTotalUsd": {
                                "type": "number",
                                "example": 0.00005
                            },
                            "usageUsd": {
                                "type": "object",
                                "properties": {
                                    "ACTOR_COMPUTE_UNITS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "DATASET_READS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "DATASET_WRITES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "KEY_VALUE_STORE_READS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "KEY_VALUE_STORE_WRITES": {
                                        "type": "number",
                                        "example": 0.00005
                                    },
                                    "KEY_VALUE_STORE_LISTS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "REQUEST_QUEUE_READS": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "REQUEST_QUEUE_WRITES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "DATA_TRANSFER_INTERNAL_GBYTES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "DATA_TRANSFER_EXTERNAL_GBYTES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "PROXY_RESIDENTIAL_TRANSFER_GBYTES": {
                                        "type": "integer",
                                        "example": 0
                                    },
                                    "PROXY_SERPS": {
                                        "type": "integer",
                                        "example": 0
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}
```
