Connessione di Event Streams a IBM Cloud Object Storage utilizzando Kubernetes Service

Ottieni il runtime Kafka Connect da eseguire in un cluster Kubernetes Service. Quindi, avviare il connettore IBM Cloud® Object Storage Per archiviare i dati dagli argomenti Kafka in Event Streams a un'istanza del servizio IBM Cloud® Object Storage.

Il connettore consuma batch di messaggi da Kafka e carica i dati dei messaggi come oggetti in un bucket del servizio Cloud Object Storage.

Passo 1. Installare i prerequisiti

Assicurati di avere installati i seguenti software e servizi:

  • Un'istanza di Event Streams - Piano Standard o Enterprise. È necessario creare le credenziali.

  • Un'istanza del servizio Cloud Object Storage con almeno un bucket.

  • Un cluster IBM Cloud® Kubernetes Service. Puoi eseguire il provisioning di un cluster gratuito per scopi di test.

    È necessario anche l'accesso CLI al cluster. Per ulteriori informazioni, vedi Configurazione della CLI e dell'API.

  • Una versione recente di Kubectl.

  • Git

Passo 2. Clona i repository kafka - connect

Clona i seguenti due repository che contengono i file richiesti:

Passo 3. Crea la tua configurazione Kafka Connect

  1. È necessario impostare questa configurazione solo una volta. Event Streams lo memorizza per un utilizzo futuro.

    Dal progetto event - streams - samples, passare a kafka-connect/IKS directory, modificare il file connect-distributed.properties e sostituire <BOOTSTRAP_SERVERS> in un posto e <APIKEY> in tre posizioni con le tue credenziali Event Streams.

    Fornire <BOOTSTRAP_SERVERS> come elenco separato da virgole. Se non sono validi, si ottiene un errore.

    Il tuo <APIKEY> appare in testo chiaro sulla tua macchina ma è segreto quando ne viene eseguito il push a IBM Cloud® Kubernetes Service.

    Kafka Connect può eseguire più nodi di lavoro per motivi di affidabilità e scalabilità. Se il tuo Kubernetes Service cluster has more than one node and you want multiple Connect workers, edit the kafka-connect.yaml file and edit the entry replicas: 1.

  2. Eseguire quindi i seguenti comandi:

    Eseguire il seguente comando per creare un segreto:

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

    Eseguire il seguente comando per creare una mappa di configurazione:

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

Passo 4. Distribuisci Kafka Connect

Applicare la configurazione nel file kafka-connect.yaml eseguendo il seguente comando:

kubectl apply -f ./kafka-connect.yaml

Passo 5. Convalida dell'esecuzione di Kafka Connect

Per convalidare che Kafka Connect è in esecuzione, inoltra la porta al servizio kafkaconnect sulla porta 8083, come nel seguente esempio:

kubectl port-forward service/kafkaconnect-service 8083

Mantenete aperto il terminale utilizzato per il port forwarding e utilizzate un altro terminale per i passi successivi.

L'API REST di Connect è quindi disponibile all'indirizzo http://localhost:8083. Per ulteriori informazioni sull'API, vedere Interfaccia RESTKafka Connect.

Quindi, ora abbiamo il runtime di Kafka Connect distribuito ed eseguito in Kubernetes Service. Quindi, configurare e avviare il connettore Object Storage.

Passo 6. Configura il file JSON cos - sink

Modificate il file cos-sink.json situato in kafka-connect-ibmcos-sink/config/ in modo che almeno le proprietà richieste siano completate con le vostre informazioni. Sebbene le proprietà di configurazione cos.object.deadline.seconds, cos.interval.seconds e cos.object.records siano elencate come opzionali, è necessario impostare almeno una di queste proprietà su un valore non predefinito.

Proprietà file cos-sink.json

Sostituire i segnaposto nel file cos-sink.json con i propri valori.

Proprietà file cos-sink.json
Proprietà file Descrizione
cos.api.key Obbligatorio. Chiave API usata per connettersi all'istanza del servizio Cloud Object Storage.
cos.bucket.location Obbligatorio. Ubicazione del bucket del servizio cloud Object Storage. Ad esempio, per un bucket regionale eu-gb o per un bucket globale eu.
cos.bucket.name Obbligatorio. Nome del bucket del servizio Cloud Object Storage in cui scrivere i dati.
cos.bucket.resiliency Obbligatorio. Resilienza del bucket Object Storage cloud. Deve essere una di: cross-region, regional o single-site.
cos.service.crn Obbligatorio. CRN per l'istanza del servizio Cloud Object Storage. Assicurarsi di immettere il CRN corretto: si tratta dell'ID istanza della risorsa che termina con doppi due punti, ad esempio, crn:v1:staging:public:cloud-object-storage:global:a/8c226dc8c8bfb9bc3431515a16957954:b25fe12c-9cf5-4ee8-8285-2c7e6ae707f6::.
cos.endpoint.visibility Facoltativo. Specifica public per la connessione al servizio Object Storage cloud su Internet pubblico. Specificare privato per connettersi da un connettore che funziona all'interno della rete IBM Cloud, ad esempio da un cluster IBM Cloud Kubernetes Service. Il valore predefinito è public.
cos.object.deadline.seconds Facoltativo. Il numero di secondi (misurato in base al tempo dell'orologio a muro dell'istanza dell'attività di connessione) tra la lettura del primo record da Kafka e la scrittura di tutti i record letti fino a quel momento in un oggetto Cloud Object Storage. Questa proprietà può essere utile in situazioni che hanno lunghe pause tra i record Kafka prodotti in un argomento. Garantisce che tutti i record ricevuti da questo connettore vengano sempre scritti in Object Storage entro il periodo di tempo specificato.
cos.object.interval.seconds Facoltativo. Il numero di secondi (misurato dai timestamp nei record Kafka ) tra la lettura del primo record da Kafka e la scrittura di tutti i record letti finora in un oggetto Cloud Object Storage.
cos.object.records Facoltativo. Il numero massimo di record Kafka da combinare in un oggetto.

Ottieni le credenziali Object Storage che utilizzano la console IBM Cloud

  1. Individua il tuo servizio Object Storage nel dashboard.
  2. Fai clic sul tuo tile del servizio.
  3. Fare clic su Credenziali del servizio.
  4. Fai clic su ** Nuova credenziale**.
  5. Completa i dettagli per la tua nuova credenziale, come ad esempio un nome e un ruolo, e fai clic su Aggiungi. Viene visualizzata una nuova credenziale nell'elenco delle credenziali.
  6. Fare clic su questa credenziale utilizzando Visualizza credenziali per visualizzare i dettagli in formato JSON.

Passo 7. Avviare il connettore con la sua configurazione

Eseguire il seguente comando per avviare il connettore Object Storage con la configurazione fornita nel passo precedente.

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

Passo 8. Monitorare il connettore

È possibile controllare il connettore andando alla seguente ubicazione.

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

Se lo stato del connettore non è in esecuzione, riavvialo.

Passo 9. Elimina il connettore

È possibile utilizzare il seguente comando per eliminare un connettore.

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