Conectando Event Streams ao IBM Cloud Object Storage usando o Kubernetes Service

Obtenha o tempo de execução Kafka Connect para ser executado em um cluster Kubernetes Service. Em seguida, inicie o IBM Cloud® Object Storage Para arquivar dados de tópicos Kafka em Event Streams em uma instância do serviço IBM Cloud® Object Storage.

O Conector consome lotes de mensagens do Kafka e faz upload dos dados da mensagem como objetos para um depósito no serviço Cloud Object Storage.

Etapa 1. Instalar os pré-requisitos

Assegure-se de que você tenha o software e os serviços a seguir instalados:

  • Uma instância do Event Streams - plano Standard ou Enterprise. É necessário criar credenciais.

  • Uma instância do serviço Cloud Object Storage com pelo menos um depósito.

  • Um cluster do IBM Cloud® Kubernetes Service. É possível provisionar um grátis para fins de teste.

    Também é necessário o acesso de CLI ao seu cluster. Para obter mais informações, consulte Configurando a CLI e a API.

  • Uma versão recente de kubectl.

  • Git

Etapa 2. Clonar os repositórios do kafka-connect

Clone os dois repositórios a seguir que contêm os arquivos necessários:

Etapa 3. Crie a configuração do Kafka Connect

  1. Deve-se definir essa configuração apenas uma vez. O Event Streams armazena-o para uso futuro.

    No projeto event-streams-samples, navegue até o arquivo kafka-connect/IKS directory, edite o arquivo connect-distributed.properties e substitua <BOOTSTRAP_SERVERS> em um local e <APIKEY> em três locais com suas credenciais do Event Streams.

    Forneça <BOOTSTRAP_SERVERS> como uma lista separada por vírgula. Se não forem válidos, você receberá um erro.

    Seu <APIKEY> aparece em texto não criptografado em sua máquina, mas é secreto quando enviado por push para o IBM Cloud® Kubernetes Service.

    O Kafka Connect pode executar múltiplos trabalhadores por motivos de confiabilidade e escalabilidade. Se o cluster do Kubernetes Service tiver mais de um nó e você desejar vários trabalhadores do Connect, edite o arquivo kafka-connect.yaml e edite a entrada replicas: 1.

  2. Em seguida, execute os comandos a seguir:

    Execute o comando a seguir para criar um segredo:

    kubectl create secret generic connect-distributed-config --from-file=connect-distributed.properties

    Execute o comando a seguir para criar um configmap:

    kubectl create configmap connect-log4j-config --from-file=connect-log4j.properties

Etapa 4. Implementar o Kafka Connect

Aplique a configuração no arquivo kafka-connect.yaml executando o seguinte comando:

kubectl apply -f ./kafka-connect.yaml

Etapa 5. Verificar se o Kafka Connect está em execução

Para verificar se o Kafka Connect está em execução, faça o encaminhamento de porta para o serviço kafkaconnect na porta 8083, como no exemplo a seguir:

kubectl port-forward service/kafkaconnect-service 8083

Mantenha o terminal que você usou para encaminhamento de porta aberto, e use outro terminal para as próximas etapas.

A API de REST do Connect está então disponível em http://localhost:8083. Se você quiser obter mais informações sobre a API, consulte Interface RESTKafka Connect.

Então, agora você tem o tempo de execução do Kafka Connect que é implementado e está em execução em Kubernetes Service. Em seguida, configure e inicie o conector Object Storage.

Etapa 6. Configurar o arquivo JSON cos-sink

Edite o arquivo cos-sink.json localizado em kafka-connect-ibmcos-sink/config/ para que, no mínimo, as suas propriedades necessárias sejam concluídas com suas informações. Embora as propriedades de configuração cos.object.deadline.seconds, cos.interval.seconds e cos.object.records estejam listadas como opcionais, é necessário definir pelo menos uma dessas propriedades com um valor não padrão.

Propriedades do arquivo cos-sink.json

Substitua os itens temporários no arquivo cos-sink.json pelos seus próprios valores.

Propriedades do arquivo cos-sink.json
Propriedades do arquivo Descrição
cos.api.key Obrigatório. Chave de API usada para se conectar à instância de serviço Cloud Object Storage.
cos.bucket.location Obrigatório. Local do depósito do serviço Cloud Object Storage. Por exemplo, para um depósito regional eu-gb, ou para um depósito global eu.
cos.bucket.name Obrigatório. Nome do depósito do serviço Cloud Object Storage para gravar dados.
cos.bucket.resiliency Obrigatório. Resiliência do depósito do Cloud Object Storage. Deve ser um de: cross-region, regional ou single-site.
cos.service.crn Obrigatório. CRN para a instância de serviço Cloud Object Storage. Certifica-se de inserir o CRN correto: é o ID da instância de recurso terminando com dois colons, por exemplo, crn:v1:staging:public:cloud-object-storage:global:a/8c226dc8c8bfb9bc3431515a16957954:b25fe12c-9cf5-4ee8-8285-2c7e6ae707f6::.
cos.endpoint.visibility Opcional. Especificar public para se conectar ao serviço Cloud Object Storage sobre a internet pública. Especifique private para se conectar a partir de um conector que é executado dentro da rede IBM Cloud, por exemplo, de um cluster IBM Cloud Kubernetes Service. O padrão é público.
cos.object.deadline.seconds Opcional. O número de segundos (como horário do relógio de parede medido para a instância Connect Task) entre a leitura do primeiro registro do Kafka e a gravação de todos os registros lidos até agora em um objeto do Cloud Object Storage. Essa propriedade pode ser útil em situações em que há longas pausas entre os registros Kafka que são produzidos para um tópico. Ele garante que quaisquer registros que sejam recebidos por esse conector estejam sempre gravados em Object Storage dentro do tempo especificado.
cos.object.interval.seconds Opcional. O número de segundos (conforme medido pelos registros de data e hora nos registros do Kafka) entre a leitura do primeiro registro do Kafka e a gravação de todos os registros lidos até agora em um objeto do Cloud Object Storage.
cos.object.records Opcional. O número máximo de registros do Kafka para combinar com um objeto.

Obter as credenciais do Object Storage que usam o console IBM Cloud

  1. Localize o seu serviço Object Storage no painel.
  2. Clique em seu bloco do serviço.
  3. Clique em Credenciais de serviço.
  4. Clique em Nova credencial.
  5. Conclua os detalhes para a sua nova credencial como um nome e uma função e clique em Incluir. Uma nova credencialaparece na lista de credenciais.
  6. Clique nesta credencial usando Visualizar Credenciais para revelar os detalhes em formato JSON.

Etapa 7. Inicie o conector com sua configuração

Execute o seguinte comando para iniciar o conector do Object Storage com a configuração que você forneceu na etapa anterior.

curl -X POST -H "Content-Type: application/json" http://localhost:8083/connectors --data "@./cos-sink.json"

Etapa 8. Monitorar o seu conector

Você pode verificar seu conector acessando o seguinte local.

http://localhost:8083/connectors/cos-sink/status

Se o estado do conector não estiver em execução, reinicie o conector.

Etapa 9. Excluir seu conector

É possível usar o comando a seguir para excluir um conector.

curl -X DELETE
http://localhost:8083/connectors/cos-sink