Praxisguide: Streame Daten von Kafka nach OpenSearch in Kubernetes

kubectl create namespace opensearch
kubectl config set-context --current --namespace=opensearch

apiVersion: apps/v1
kind: Deployment
metadata:
  name: opensearch
  namespace: opensearch
spec:
  replicas: 1
  selector:
    matchLabels:
      app: opensearch
  template:
    metadata:
      labels:
        app: opensearch
    spec:
      containers:
      - name: opensearch
        image: opensearchproject/opensearch:latest
        ports:
        - containerPort: 9200
        env:
        - name: discovery.type
          value: "single-node"
        - name: plugins.security.disabled
          value: "true"
---
apiVersion: v1
kind: Service
metadata:
  name: opensearch-service
  namespace: opensearch
spec:
  selector:
    app: opensearch
  ports:
  - port: 9200
    targetPort: 9200

kubectl apply -f opensearch.yaml

kubectl get pods -w

kubectl port-forward svc/opensearch-service 9200:9200

curl http://localhost:9200/ | jq

curl -X PUT "http://localhost:9200/db-transactions" | jq

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: connect-cluster
  namespace: opensearch
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.7.0
  replicas: 1
  bootstrapServers: kafka-kafka-bootstrap.kafka.svc.cluster.local:9092
  config:
    group.id: connect-cluster
    # Wir benötigen den JSON Converter mit Schema-Support
    key.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: true
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: true
  # Mein Image mit integriertem OpenSearch-Connector
  # Anleitung, wie du dieses Image selbst erstellen kannst, findest du weiter unten
  image: ghcr.io/kafka-trainings/example-project/opensearch-connector:latest

kubectl apply -f kafka-connect.yaml
# Es dauert bis zu 2 Minuten
kubectl get pods -w

# Builder image
FROM gradle:7.6 as builder
# Wir nutzen eine ältere Version, damit der Blogpost nicht so schnell veraltet
RUN wget https://github.com/Aiven-Open/opensearch-connector-for-apache-kafka/archive/refs/tags/v3.1.1.tar.gz
RUN tar -xzf v3.1.1.tar.gz && \
    cd opensearch-connector-for-apache-kafka-3.1.1 && \
    ./gradlew installDist

# Image für Kafka Connect
# Wir verwenden hier Kafka 3.7.0. Nutze bitte die neueste Version!
FROM quay.io/strimzi/kafka:latest-kafka-3.7.0
USER root:root
# Connector kopieren
COPY --from=builder /home/gradle/opensearch-connector-for-apache-kafka-3.1.1/build/install/opensearch-connector-for-apache-kafka/ /opt/kafka/plugins/
USER 1001

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: opensearch-sink-connector
  labels:
    strimzi.io/cluster: connect-cluster
spec:
  class: io.aiven.kafka.connect.opensearch.OpensearchSinkConnector
  tasksMax: 1
  config:
    topics: "db-transactions"
    key.ignore: true
    connection.url: "http://opensearch-service.opensearch.svc.cluster.local:9200"

kubectl apply -f opensearch-sink-connector.yaml
kubectl describe kafkaconnector opensearch-sink-connector

{
"type": "struct",
"fields": [
    {
    "field": "id",
    "type": "int32"
    },
    {
    "field": "amount",
    "type": "int32"
    },
    {
    "field": "name",
    "type": "string"
    }
]
}

{"id": 1, "amount": 100, "name": "John Doe"}
{"id": 2, "amount": 200, "name": "Jane Doe"}
{"id": 3, "amount": 300, "name": "John Smith"}

{"schema": {"type": "struct", "fields": [{"field": "id", "type": "int32"}, {"field": "amount", "type": "int32"}, {"field": "name", "type": "string"}]}, "payload": {"id": 1, "amount": 100, "name": "John Doe"}}

kubectl exec -it kafka-kafka-0 -- \
bin/kafka-console-producer.sh \
--bootstrap-server kafka-kafka-bootstrap.kafka.svc.cluster.local:9092 \
--topic db-transactions

{"schema": {"type": "struct", "fields": [{"field": "id", "type": "int32"}, {"field": "amount", "type": "int32"}, {"field": "name", "type": "string"}]}, "payload": {"id": 1, "amount": 100, "name": "John Doe"}}
{"schema": {"type": "struct", "fields": [{"field": "id", "type": "int32"}, {"field": "amount", "type": "int32"}, {"field": "name", "type": "string"}]}, "payload": {"id": 2, "amount": 200, "name": "Jane Doe"}}
{"schema": {"type": "struct", "fields": [{"field": "id", "type": "int32"}, {"field": "amount", "type": "int32"}, {"field": "name", "type": "string"}]}, "payload": {"id": 3, "amount": 300, "name": "John Smith"}}

curl -X GET "http://localhost:9200/db-transactions/_search?q=id:1" | jq