La tabella seguente elenca i tipi di connettori Kafka Connect supportati da Managed Service per Apache Kafka. Puoi utilizzare questi connettori per integrare Apache Kafka con le tue applicazioni e con altri servizi Google Cloud.
| Connettore | Descrizione | Casi d'uso |
|---|---|---|
| Sink BigQuery | Trasmette i dati dagli argomenti Kafka a una tabella BigQuery. | Data warehousing, analisi |
| Origine Cloud SQL per PostgreSQL | Legge le modifiche a livello di riga da un database Cloud SQL per PostgreSQL e le scrive in Kafka. | Integrazione del database, monitoraggio e risposta agli eventi del database |
| Sink di Cloud Storage | Trasmette i dati dagli argomenti Kafka a un bucket Cloud Storage. | Importazione di data lake, archiviazione dei dati |
| Origine PostgreSQL generica | Legge le modifiche a livello di riga da un database PostgreSQL e le scrive su Kafka. | Integrazione del database, monitoraggio e risposta agli eventi del database |
| Checkpoint MirrorMaker 2.0 | Copia gli offset dei consumer da un cluster Kafka a un altro. | Replica dei dati, ripristino di emergenza, migrazione dei dati |
| Heartbeat MirrorMaker 2.0 | Genera messaggi di heartbeat periodici e li scrive in un argomento | Monitoraggio dell'integrità del cluster, della connessione di rete o della latenza di replica end-to-end tra i cluster. |
| Origine MirrorMaker 2.0 | Replica gli argomenti e i dati da un cluster Kafka a un altro. | Replica dei dati, ripristino di emergenza, migrazione dei dati |
| Sink Pub/Sub | Trasmette i dati dagli argomenti Kafka a un argomento Pub/Sub. | Integrazione dei servizi, notifiche in tempo reale |
| Origine Pub/Sub | Trasmette i messaggi da una sottoscrizione Pub/Sub a un argomento Kafka. | Importazione dati in tempo reale, architetture basate su eventi |
Utenti che hanno completato una conversione
I convertitori sono responsabili della serializzazione e della deserializzazione dei dati dei record Kafka. Traducono tra il formato di byte non elaborati presente negli argomenti Kafka e la rappresentazione interna dei dati strutturati utilizzata da Kafka Connect.
Per i connettori sink, i convertitori deserializzano i dati dal formato di rete dell'argomento al formato dei dati interno di Kafka Connect, che il connettore utilizza per scrivere nel sistema di destinazione.
Per i connettori di origine, i convertitori serializzano i dati dal formato dei dati interno di Kafka Connect al formato di rete specificato per l'argomento Kafka.
I convertitori assicurano che il connettore legga o scriva i record Kafka in un formato compatibile con il sistema esterno.
Quando configuri un connettore, imposta le seguenti proprietà:
Convertitore di chiavi (
key.converter): il convertitore da utilizzare durante la serializzazione e la deserializzazione delle chiavi dei record Kafka.Convertitore di valori (
value.converter): il convertitore da utilizzare durante la serializzazione e la deserializzazione dei valori dei record Kafka.
Se non specifichi un convertitore, il tipo di convertitore predefinito è org.apache.kafka.connect.converters.ByteArrayConverter, che trasmette i dati nel formato di byte non elaborati.
Convertitori supportati
Managed Service per Apache Kafka supporta i seguenti convertitori integrati:
| Utente che ha completato una conversione | Formato |
|---|---|
io.confluent.connect.avro.AvroConverter |
Apache Avro |
org.apache.kafka.connect.converters.BooleanConverter |
Booleano |
org.apache.kafka.connect.converters.ByteArrayConverter |
Array di byte Il tipo di convertitore predefinito. Conserva i contenuti esatti dei messaggi tra due sistemi. |
org.apache.kafka.connect.converters.DoubleConverter |
Doppio |
org.apache.kafka.connect.converters.FloatConverter |
Numero in virgola mobile |
org.apache.kafka.connect.converters.IntegerConverter |
Numero intero |
org.apache.kafka.connect.json.JsonConverter |
JSON Per i dati JSON senza uno schema, imposta anche |
org.apache.kafka.connect.converters.LongConverter |
Lungo |
org.apache.kafka.connect.converters.ShortConverter |
Breve |
org.apache.kafka.connect.storage.StringConverter |
Stringa |
La scelta del convertitore dipende dal tipo di connettore e dai dati che archivi in Kafka. Per saperne di più, consulta la documentazione del connettore specifico.
Tasks
Un connettore trasferisce i dati creando una o più attività che operano in parallelo. Per impostare un limite massimo per il numero di attività create da un connettore, imposta la proprietà di configurazione tasks.max del connettore. Il connettore potrebbe creare un numero di attività inferiore a questo valore.
L'aumento del valore di tasks.max può migliorare il throughput, ma anche aumentare il consumo di risorse (CPU e memoria). Il valore ottimale dipende dal workload e dalle risorse allocate ai worker del cluster Connect. Per i connettori sink, anche il numero di partizioni degli argomenti Kafka può influire sul parallelismo.
Policy di riavvio attività
Puoi impostare la policy di riavvio delle attività di un connettore, che determina il comportamento in caso di errore. I connettori supportano le seguenti policy:
Non riavviare mai. Il connettore non riavvia le attività non riuscite. Questa policy è il comportamento predefinito. È utile per il debug o in situazioni in cui è necessario un intervento manuale dopo un errore.
Riavvia con backoff esponenziale. Il connettore riavvia un'attività non riuscita dopo un ritardo (chiamato periodo di backoff). Il ritardo aumenta in modo esponenziale con ogni errore successivo. Questa policy è consigliata per la maggior parte dei workload di produzione.
Se utilizzi la policy di backoff esponenziale, imposta anche i valori per il backoff minimo e massimo. Il backoff minimo deve essere maggiore di 60 secondi, mentre il backoff massimo deve essere inferiore a 7200 secondi.
Trasformazioni e predicati
Managed Service per Apache Kafka supporta le trasformazioni e i predicati predefiniti di Kafka Connect.
Le trasformazioni consentono di modificare i singoli messaggi prima che vengano inviati a Managed Service per Apache Kafka (per i connettori di origine) o al sistema esterno (per i connettori sink). Potresti utilizzare una trasformazione per mascherare dati sensibili, aggiungere timestamp o rinominare campi.
I predicati consentono di filtrare i dati in base a condizioni specifiche, determinando a quali messaggi si applica una trasformazione in base alle proprietà dei messaggi.
Ad esempio, per configurare un connettore sink in modo che ignori i messaggi che contengono una chiave di intestazione DoNotProcess, aggiungi la seguente configurazione:
transforms=dropMessage
transforms.dropMessage.type=org.apache.kafka.connect.transforms.Filter
transforms.dropMessage.predicate=hasKey
predicates=hasKey
predicates.hasKey.type=org.apache.kafka.connect.transforms.predicates.HasHeaderKey
predicates.hasKey.name=DoNotProcess
Questa configurazione prevede quanto segue:
Configura un predicato denominato
hasKeydi tipoorg.apache.kafka.connect.transforms.predicates.HasHeaderKey. Questo predicato corrisponde a tutti i messaggi che contengono un'intestazione con la chiaveDoNotProcess.Configura una trasformazione denominata
dropMessagedi tipoorg.apache.kafka.connect.transforms.Filter. Questa trasformazione elimina tutti i messaggi che corrispondono al predicato configurato.Collega la trasformazione al predicato
hasKey. In questo modo, solo i messaggi con la chiave di intestazioneDoNotProcessvengono ignorati dalla trasformazione.