Prima di iniziare Creazione di un connettore sorgente Debezium

Connettore sorgente Debezium con provider di configurazione

Questo esempio mostra come utilizzare il plug-in del connettore Debezium MySQL con un database Amazon Aurora compatibile con MySQL come origine. In questo esempio, abbiamo anche impostato il provider di Config open sourceAWS Secrets Manager per esternalizzare le credenziali del database inAWS Secrets Manager. Per ulteriori informazioni sui provider di configurazione, vedereEsternalizzazione di informazioni sensibili utilizzando i provider di configurazione.

Importante

Il plug-in del connettore Debezium MySQL supporta solo un'operazione e non funziona con la modalità di capacità con scalabilità automatica per Amazon MSK Connect. Dovresti invece utilizzare la modalità di capacità fornita e impostarla comeworkerCount uguale a uno nella configurazione del connettore. Per ulteriori informazioni sulle modalità di capacità di MSK Connect, vedereCapacità del connettore.

Prima di iniziare

Il connettore deve essere in grado di accedere a Internet in modo da poter interagire con servizi esterni all'utenteAmazon Virtual Private Cloud.AWS Secrets Manager I passaggi in questa sezione consentono di completare le seguenti attività per abilitare l'accesso a Internet.

Configura una sottorete pubblica che ospita un gateway NAT e indirizza il traffico verso un gateway Internet nel tuo VPC.
Crea un percorso predefinito che indirizza il traffico di sottorete privato verso il gateway NAT.

Per ulteriori informazioni, consulta Abilitazione dell'accesso a Internet per Amazon MSK Connect.

Prerequisiti

Prima di poter abilitare l'accesso a Internet, devi disporre dei seguenti elementi:

L'ID delAmazon Virtual Private Cloud (VPC) associato al cluster. Ad esempio, vpc-123456ab.
Gli ID delle sottoreti private nel VPC. Ad esempio, subnet-a1b2c3de, subnet-f4g5h6ij, ecc. È necessario configurare il connettore con sottoreti private.

Per abilitare l'accesso a Internet per il connettore

Aprire la console Amazon Virtual Private Cloud all'indirizzo https://console.aws.amazon.com/vpc/.
Crea una sottorete pubblica per il tuo gateway NAT con un nome descrittivo e annota l'ID della sottorete. Per istruzioni dettagliate, consulta Creazione di una sottorete nel VPC.
Creazione di un gateway Internet in modo che il VPC possa comunicare con Internet e annotare l'ID del gateway. Collegare il gateway internet al VPC. Per istruzioni, vedere Creazione e collegamento di un Internet Gateway.
Fornisci un gateway NAT pubblico in modo che gli host delle tue sottoreti private possano raggiungere la tua sottorete pubblica. Quando crei il gateway NAT, seleziona la sottorete pubblica creata in precedenza. Per istruzioni, consulta Creazione di un gateway NAT.

Configura le tabelle dei percorsi. È necessario disporre di due tabelle di percorso in totale per completare questa configurazione. Dovresti già avere una tabella dei percorsi principali che è stata creata automaticamente contemporaneamente al tuo VPC. In questo passaggio si crea una tabella di routing aggiuntiva per la sottorete pubblica.

Utilizza le seguenti impostazioni per modificare la tabella dei percorsi principali del tuo VPC in modo che le tue sottoreti private indirizzino il traffico verso il tuo gateway NAT. Per istruzioni, vedere Lavorare con le tabelle dei percorsi nella Guida per l'Amazon Virtual Private Cloudutente.

Tabella dei percorsi MSKC privata
Proprietà	Value (Valore)
Name tag (Tag nome)	Ti consigliamo di assegnare a questa tabella dei percorsi un tag descrittivo per aiutarti a identificarla. Ad esempio, Private MSKC.
Sottoreti associate	Sottoreti private tue sottoreti private
Un percorso per abilitare l'accesso a Internet per MSK Connect	Destinazione: 0.0/0 0 0 0 0 0 0 0 0 0 Obiettivo: l'ID del tuo gateway NAT. Ad esempio, nat-12a345bc6789efg1h.
Un percorso locale per il traffico interno	Destinazione: 10.0.0/16 16. 0/16 16. 0/16 16. Questo valore può variare a seconda del blocco CIDR del VPC. Destinazione: locale

Segui le istruzioni in Crea una tabella dei percorsi personalizzata per creare una tabella dei percorsi per la tua sottorete pubblica. Quando crei la tabella, inserisci un nome descrittivo nel campo del tag Nome per aiutarti a identificare a quale sottorete è associata la tabella. Ad esempio, Public MSKC.

Configura la tabella dei percorsi MSKC pubblica utilizzando le seguenti impostazioni.

Proprietà	Value (Valore)
Name tag (Tag nome)	MSKC pubblico o un nome descrittivo diverso a tua scelta
Sottoreti associate	La tua sottorete pubblica con gateway NAT
Un percorso per abilitare l'accesso a Internet per MSK Connect	Destinazione: 0.0/0 0 0 0 0 0 0 0 0 0 Obiettivo: l'ID del tuo gateway Internet. Ad esempio, igw-1a234bc5.
Un percorso locale per il traffico interno	Destinazione: 10.0.0/16 16. 0/16 16. 0/16 16. Questo valore può variare a seconda del blocco CIDR del VPC. Destinazione: locale

Ora che hai abilitato l'accesso a Internet per Amazon MSK Connect, sei pronto per creare un connettore.

Creazione di un connettore sorgente Debezium

Creazione di un plugin personalizzato personalizzato
1. Scarica il plug-in del connettore MySQL per l'ultima versione stabile dal sito Debezium. Prendi nota della versione di rilascio di Debezium che scarichi (versione 2.x o la vecchia serie 1.x). Più avanti in questa procedura, creerai un connettore basato sulla tua versione di Debezium.
2. Scarica ed estrai il provider di Config diAWS Secrets Manager.
3. Inserisci i seguenti archivi nella stessa cartella:
  - Ladebezium-connector-mysql cartella
  - Lajcusten-border-kafka-config-provider-aws-0.1.1 cartella
4. Comprimi la directory creata nel passaggio precedente in un file ZIP, quindi carica il file ZIP in un bucket S3. Per istruzioni, consulta Caricare oggetti nella Guida per l'utente di Amazon S3.
5. Copiare il seguente JSON e incollarlo in un file. Ad esempio, debezium-source-custom-plugin.json. Sostituisci <example-custom-plugin-name > con il nome che vuoi che abbia il plugin, <arn-of-your-s 3-bucket> con l'ARN del bucket S3 in cui hai caricato il file ZIP e<file-key-of-ZIP-object> con la chiave del file dell'oggetto ZIP che hai caricato su S3.
```
{
    "name": "<example-custom-plugin-name>",
    "contentType": "ZIP",
    "location": {
        "s3Location": {
            "bucketArn": "<arn-of-your-s3-bucket>",
            "fileKey": "<file-key-of-ZIP-object>"
        }
    }
}
```
6. Esegui il seguenteAWS CLI comando dalla cartella in cui hai salvato il file JSON per creare un plugin.
```
aws kafkaconnect create-custom-plugin --cli-input-json file://<debezium-source-custom-plugin.json>
```
  Verrà visualizzato un output simile all'esempio seguente.
```
{
    "CustomPluginArn": "arn:aws:kafkaconnect:us-east-1:012345678901:custom-plugin/example-custom-plugin-name/abcd1234-a0b0-1234-c1-12345678abcd-1",
    "CustomPluginState": "CREATING",
    "Name": "example-custom-plugin-name",
    "Revision": 1
}
```
7. Utilizzare il seguente comando per verificare lo stato del plugin. Lo stato dovrebbe cambiare daCREATING aACTIVE. Sostituire il segnapeti ARN con l'ARN che si trova nell'output del comando precedente.
```
aws kafkaconnect describe-custom-plugin --custom-plugin-arn "<arn-of-your-custom-plugin>"
```
ConfiguraAWS Secrets Manager e crea un segreto per le credenziali del tuo database
1. Apri la console di Secrets Manager all'indirizzo https://console.aws.amazon.com/secretsmanager/.
2. Crea un nuovo segreto per archiviare le credenziali di accesso al database. Per istruzioni, consulta Creare un segreto nella Guida per l'AWS Secrets Managerutente.
3. Copia l'ARN del tuo segreto.
4. Aggiungi le autorizzazioni di Secrets Manager dalla seguente politica di esempio alla tuaRuolo di esecuzione del servizio. Sostituisci <arn:aws:secretsmanager:us-east- 1:123456789000:secret:MySecret -1234> con l'ARN del tuo segreto.
```
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetResourcePolicy",
        "secretsmanager:GetSecretValue",
        "secretsmanager:DescribeSecret",
        "secretsmanager:ListSecretVersionIds"
      ],
      "Resource": [
        "<arn:aws:secretsmanager:us-east-1:123456789000:secret:MySecret-1234>"
      ]
    }
  ]
}
```
  Per istruzioni su come aggiungere autorizzazioni IAM, vedere Aggiunta e rimozione di autorizzazioni per identità IAM nella Guida per l'utente IAM.
Crea una configurazione di lavoro personalizzata con informazioni sul tuo provider di configurazione
1. Copia le seguenti proprietà di configurazione del lavoratore in un file, sostituendo le stringhe segnaposto con valori che corrispondono al tuo scenario. Per ulteriori informazioni sulle proprietà di configurazione del provider di Config diAWS Secrets Manager, consulta SecretsManagerConfigProviderla documentazione del plugin.
```
key.converter=<org.apache.kafka.connect.storage.StringConverter>
value.converter=<org.apache.kafka.connect.storage.StringConverter>
config.providers.secretManager.class=com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProvider
config.providers=secretManager
config.providers.secretManager.param.aws.region=<us-east-1>
```
2. Esegui ilAWS CLI comando seguente per creare la tua configurazione di lavoro personalizzata.
  
  Sostituisci i valori seguenti:
  - <my-worker-config-name > - un nome descrittivo per la configurazione di lavoro personalizzata
  - <encoded-properties-file-content -string> - una versione con codifica base64 delle proprietà di testo normale che hai copiato nel passaggio precedente
```
aws kafkaconnect create-worker-configuration --name <my-worker-config-name> --properties-file-content <encoded-properties-file-content-string>
```

Crea un connettore

Copia il seguente JSON che corrisponde alla tua versione di Debezium (2.x o 1.x) e incollalo in un nuovo file. Sostituisci<placeholder> le stringhe con valori che corrispondono al tuo scenario. Per informazioni su come configurare un ruolo di esecuzione del servizio, vedereRuoli e politiche IAM per MSK Connect.

Nota che la configurazione utilizza variabili come${secretManager:MySecret-1234:dbusername} anziché testo normale per specificare le credenziali del database. SostituisciMySecret-1234 con il nome del tuo segreto e poi includi il nome della chiave che desideri recuperare. È inoltre necessario sostituire<arn-of-config-provider-worker-configuration> con l'ARN della configurazione di lavoro personalizzata.

Debezium 2.x

Per le versioni di Debezium 2.x, copiate il seguente JSON e incollatelo in un nuovo file. Sostituisci <placeholder>le stringhe con valori che corrispondono al tuo scenario.


{
	"connectorConfiguration": {
		"connector.class": "io.debezium.connector.mysql.MySqlConnector",
		"tasks.max": "1",
		"database.hostname": "<aurora-database-writer-instance-endpoint>",
		"database.port": "3306",
		"database.user": "<${secretManager:MySecret-1234:dbusername}>",
		"database.password": "<${secretManager:MySecret-1234:dbpassword}>",
		"database.server.id": "123456",
		"database.include.list": "<list-of-databases-hosted-by-specified-server>",
		"topic.prefix": "<logical-name-of-database-server>",
		"schema.history.internal.kafka.topic": "<kafka-topic-used-by-debezium-to-track-schema-changes>",
		"schema.history.internal.kafka.bootstrap.servers": "<cluster-bootstrap-servers-string>",
		"schema.history.internal.consumer.security.protocol": "SASL_SSL",
		"schema.history.internal.consumer.sasl.mechanism": "AWS_MSK_IAM",
		"schema.history.internal.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
		"schema.history.internal.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
		"schema.history.internal.producer.security.protocol": "SASL_SSL",
		"schema.history.internal.producer.sasl.mechanism": "AWS_MSK_IAM",
		"schema.history.internal.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
		"schema.history.internal.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
		"include.schema.changes": "true"
	},
	"connectorName": "example-Debezium-source-connector",
	"kafkaCluster": {
		"apacheKafkaCluster": {
			"bootstrapServers": "<cluster-bootstrap-servers-string>",
			"vpc": {
				"subnets": [
					"<cluster-subnet-1>",
					"<cluster-subnet-2>",
					"<cluster-subnet-3>"
				],
				"securityGroups": ["<id-of-cluster-security-group>"]
			}
		}
	},
	"capacity": {
		"provisionedCapacity": {
			"mcuCount": 2,
			"workerCount": 1
		}
	},
	"kafkaConnectVersion": "2.7.1",
	"serviceExecutionRoleArn": "<arn-of-service-execution-role-that-msk-connect-can-assume>",
	"plugins": [{
		"customPlugin": {
			"customPluginArn": "<arn-of-msk-connect-plugin-that-contains-connector-code>",
			"revision": 1
		}
	}],
	"kafkaClusterEncryptionInTransit": {
		"encryptionType": "TLS"
	},
	"kafkaClusterClientAuthentication": {
		"authenticationType": "IAM"
	},
	"workerConfiguration": {
		"workerConfigurationArn": "<arn-of-config-provider-worker-configuration>",
		"revision": 1
	}
}

Debezium 1.x

Per le versioni di Debezium 1.x, copia il seguente JSON e incollalo in un nuovo file. Sostituisci <placeholder>le stringhe con valori che corrispondono al tuo scenario.


{
	"connectorConfiguration": {
		"connector.class": "io.debezium.connector.mysql.MySqlConnector",
		"tasks.max": "1",
		"database.hostname": "<aurora-database-writer-instance-endpoint>",
		"database.port": "3306",
		"database.user": "<${secretManager:MySecret-1234:dbusername}>",
		"database.password": "<${secretManager:MySecret-1234:dbpassword}>",
		"database.server.id": "123456",
		"database.server.name": "<logical-name-of-database-server>",
		"database.include.list": "<list-of-databases-hosted-by-specified-server>",
		"database.history.kafka.topic": "<kafka-topic-used-by-debezium-to-track-schema-changes>",
		"database.history.kafka.bootstrap.servers": "<cluster-bootstrap-servers-string>",
		"database.history.consumer.security.protocol": "SASL_SSL",
		"database.history.consumer.sasl.mechanism": "AWS_MSK_IAM",
		"database.history.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
		"database.history.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
		"database.history.producer.security.protocol": "SASL_SSL",
		"database.history.producer.sasl.mechanism": "AWS_MSK_IAM",
		"database.history.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
		"database.history.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
		"include.schema.changes": "true"
	},
	"connectorName": "example-Debezium-source-connector",
	"kafkaCluster": {
		"apacheKafkaCluster": {
			"bootstrapServers": "<cluster-bootstrap-servers-string>",
			"vpc": {
				"subnets": [
					"<cluster-subnet-1>",
					"<cluster-subnet-2>",
					"<cluster-subnet-3>"
				],
				"securityGroups": ["<id-of-cluster-security-group>"]
			}
		}
	},
	"capacity": {
		"provisionedCapacity": {
			"mcuCount": 2,
			"workerCount": 1
		}
	},
	"kafkaConnectVersion": "2.7.1",
	"serviceExecutionRoleArn": "<arn-of-service-execution-role-that-msk-connect-can-assume>",
	"plugins": [{
		"customPlugin": {
			"customPluginArn": "<arn-of-msk-connect-plugin-that-contains-connector-code>",
			"revision": 1
		}
	}],
	"kafkaClusterEncryptionInTransit": {
		"encryptionType": "TLS"
	},
	"kafkaClusterClientAuthentication": {
		"authenticationType": "IAM"
	},
	"workerConfiguration": {
		"workerConfigurationArn": "<arn-of-config-provider-worker-configuration>",
		"revision": 1
	}
}

Esegui ilAWS CLI comando seguente nella cartella in cui hai salvato il file JSON nel passaggio precedente.


aws kafkaconnect create-connector --cli-input-json file://connector-info.json

Di seguito è riportato un esempio dell'output che si ottiene quando si esegue correttamente il comando.


{
    "ConnectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2", 
    "ConnectorState": "CREATING", 
    "ConnectorName": "example-Debezium-source-connector"
}

Per un esempio di connettore Debezium con passaggi dettagliati, consulta Introduzione ad Amazon MSK Connect - Streaming di dati da e verso i cluster Apache Kafka utilizzando connettori gestiti.

Avvertimento JavaScript è disabilitato o non è disponibile nel tuo browser.

Per usare la documentazione AWS, JavaScript deve essere abilitato. Consulta le pagine della guida del browser per le istruzioni.

Convenzioni dei documenti

Connettore per lavandino Amazon S3

Best practice

Connettore sorgente Debezium con provider di configurazione

Importante

Prima di iniziare

Per abilitare l'accesso a Internet per il connettore

Creazione di un connettore sorgente Debezium

Creazione di un plugin personalizzato personalizzato

ConfiguraAWS Secrets Manager e crea un segreto per le credenziali del tuo database

Crea una configurazione di lavoro personalizzata con informazioni sul tuo provider di configurazione

Crea un connettore