Apache Kafka

Die Apache Kafka (Kafka) -Aktion sendet Nachrichten direkt an Ihr Amazon Managed Streaming for Apache Kafka (Amazon MSK), Apache Kafka-Cluster, die von Drittanbietern wie Confluent Cloud verwaltet werden, oder an selbstverwaltete Apache Kafka-Cluster zur Datenanalyse und -visualisierung.

Anmerkung

Dieses Thema setzt Vertrautheit mit der Apache Kafka-Plattform und verwandten Konzepten voraus. Weitere Informationen zu Apache Kafka finden Sie unter Apache Kafka. MSK Serverless wird nicht unterstützt. Serverlose MSK-Cluster können nur über die IAM-Authentifizierung ausgeführt werden, die die Apache Kafka-Regelaktion derzeit nicht unterstützt.

Voraussetzungen

Diese Regelaktion hat die folgenden Anforderungen:

Eine IAM-Rolle, die die Ausführung derec2:CreateNetworkInterface,,,, ec2:DescribeNetworkInterfaces ec2:CreateNetworkInterfacePermission ec2:DeleteNetworkInterfaceec2:DescribeSubnets, ec2:DescribeVpcs und -Operationen übernehmen AWS IoT kann. ec2:DescribeVpcAttribute ec2:DescribeSecurityGroups Diese Rolle erstellt und verwaltet ENIs zu Ihrer Amazon Virtual Private Cloud, um Ihren Kafka-Broker zu erreichen. Weitere Informationen finden Sie unter Gewähren Sie einer AWS IoT Regel den Zugriff, den sie benötigt.

In der AWS IoT Konsole können Sie eine Rolle auswählen oder erstellen, um die Ausführung dieser Regelaktion AWS IoT Core zu ermöglichen.

Weitere Informationen zu Netzwerkschnittstellen finden Sie unter Elastic-Network-Schnittstellen im Amazon-EC2-Benutzerhandbuch.

Die Richtlinie, die der von Ihnen angegebenen Rolle zugewiesen ist, sollte wie im folgenden Beispiel aussehen.
```
{
    "Version": "2012-10-17",
    "Statement": [
    {
        "Effect": "Allow",
        "Action": [
            "ec2:CreateNetworkInterface",
            "ec2:DescribeNetworkInterfaces",
            "ec2:CreateNetworkInterfacePermission",
            "ec2:DeleteNetworkInterface",
            "ec2:DescribeSubnets",
            "ec2:DescribeVpcs",
            "ec2:DescribeVpcAttribute",
            "ec2:DescribeSecurityGroups"
            ],
            "Resource": "*"
        }
    ]
}                     
```

Wenn Sie die Anmeldeinformationen speichern AWS Secrets Manager , die für die Verbindung mit Ihrem Kafka-Broker erforderlich sind, müssen Sie eine IAM-Rolle erstellen, die die Ausführung der secretsmanager:GetSecretValue AND-Operationen übernehmen AWS IoT Core kann. secretsmanager:DescribeSecret

Die Richtlinie, die der von Ihnen angegebenen Rolle zugewiesen ist, sollte wie im folgenden Beispiel aussehen.


{
    "Version": "2012-10-17",
    "Statement": [
    {
        "Effect": "Allow",
        "Action": [
            "secretsmanager:GetSecretValue",
            "secretsmanager:DescribeSecret"
            ],
            "Resource": [
                "arn:aws:secretsmanager:region:123456789012:secret:kafka_client_truststore-*",
                "arn:aws:secretsmanager:region:123456789012:secret:kafka_keytab-*"
            ]
        }
    ]
}

Sie können Ihre Apache Kafka-Cluster in Amazon Virtual Private Cloud (Amazon VPC) ausführen. Sie müssen ein Amazon VPC-Ziel erstellen und ein NAT-Gateway in Ihren Subnetzen verwenden, um Nachrichten von AWS IoT an einen öffentlichen Kafka-Cluster weiterzuleiten. Die AWS IoT Regel-Engine erstellt in jedem der im VPC-Ziel aufgeführten Subnetze eine Netzwerkschnittstelle, um den Verkehr direkt an die VPC weiterzuleiten. Wenn Sie ein VPC-Ziel erstellen, erstellt die AWS IoT Regel-Engine automatisch eine VPC-Regelaktion. Weitere Informationen zu VPC-Regelaktionen finden Sie unter Virtual Private Cloud (VPC) -Ziele.
Wenn Sie einen vom Kunden verwalteten AWS KMS key (KMS-Schlüssel) verwenden, um Daten im Ruhezustand zu verschlüsseln, muss der Dienst die Erlaubnis haben, den KMS-Schlüssel im Namen des Anrufers zu verwenden. Weitere Informationen finden Sie unter Amazon-MSK-Verschlüsselung im Entwicklerhandbuch für Amazon Managed Streaming for Apache Kafka.

Parameter

Wenn Sie mit dieser Aktion eine AWS IoT Regel erstellen, müssen Sie die folgenden Informationen angeben:

destinationArn

Der Amazon-Ressourcenname (ARN) des VPC-Ziels. Weitere Informationen zum Erstellen eines VPC-Ziels finden Sie unter Virtual Private Cloud (VPC) -Ziele.

Thema

Das Kafka-Thema für Nachrichten, die an den Kafka-Broker gesendet werden sollen.

Sie können dieses Feld durch eine Ersatzvorlage ersetzen. Weitere Informationen finden Sie unter Ersetzungsvorlagen.

Schlüssel (optional)

Der Kafka-Nachrichtenschlüssel.

Sie können dieses Feld durch eine Ersatzvorlage ersetzen. Weitere Informationen finden Sie unter Ersetzungsvorlagen.

Überschriften (optional)

Die Liste der Kafka-Header, die Sie festlegen. Jeder Header ist ein Schlüssel-Wert-Paar, das Sie beim Erstellen einer Kafka-Aktion angeben können. Sie können diese Header verwenden, um Daten von IoT-Clients an Downstream-Kafka-Cluster weiterzuleiten, ohne Ihre Nachrichtennutzlast zu ändern.

Sie können dieses Feld durch eine Ersatzvorlage ersetzen. Um zu verstehen, wie die Funktion einer Inline-Regel als Ersatzvorlage in den Header von Kafka Action übergeben wird, siehe Beispiele. Weitere Informationen finden Sie unter Ersetzungsvorlagen.

Anmerkung

Header im Binärformat werden nicht unterstützt.

Partition (optional)

Die Kafka-Nachrichten-Partition.

Sie können dieses Feld durch eine Ersatzvorlage ersetzen. Weitere Informationen finden Sie unter Ersetzungsvorlagen.

clientProperties

Ein Objekt, das die Eigenschaften des Apache Kafka-Producer-Clients definiert.

acks (optional)

Die Anzahl der Bestätigungen, die der Hersteller beim Server erhalten haben muss, bevor eine Anfrage als abgeschlossen betrachtet werden kann.

Wenn Sie 0 als Wert angeben, wartet der Producer nicht auf eine Bestätigung vom Server. Wenn der Server die Nachricht nicht empfängt, versucht der Producer nicht erneut, die Nachricht zu senden.

Zulässige Werte: -1, 0, 1, all. Der Standardwert ist 1.

bootstrap.servers

Eine Liste von Host- und Port-Paaren (z. B.host1:port1, host2:port2), die verwendet wurden, um die erste Verbindung zu Ihrem Kafka-Cluster herzustellen.

compression.type (optional)

Der Komprimierungstyp für alle vom Hersteller generierten Daten.

Zulässige Werte: none, gzip, snappy, lz4, zstd. Der Standardwert ist none.

security.protocol

Das Sicherheitsprotokoll, das für die Verbindung mit Ihrem Kafka-Broker verwendet wird.

Zulässige Werte: SSL, SASL_SSL. Der Standardwert ist SSL.

key.serializer

Gibt an, wie die Schlüsselobjekte, die Sie mit ProducerRecord bereitstellen, in Bytes umgewandelt werden sollen.

Zulässiger Wert: StringSerializer.

value.serializer

Gibt an, wie Wertobjekte, die Sie mit dem ProducerRecord bereitstellen, in Bytes umgewandelt werden sollen.

Zulässiger Wert: ByteBufferSerializer.

ssl.truststore

Die Truststore-Datei im Base64-Format oder der Speicherort der Truststore-Datei in AWS Secrets Manager. Dieser Wert ist nicht erforderlich, wenn Ihr Truststore von den Amazon-Zertifizierungsstellen (CA) als vertrauenswürdig eingestuft wird.

Dieses Feld unterstützt Substitutionsvorlagen. Wenn Sie Secrets Manager verwenden, um die Anmeldeinformationen zu speichern, die für die Verbindung mit Ihrem Kafka-Broker erforderlich sind, können Sie die get_secret SQL-Funktion verwenden, um den Wert für dieses Feld abzurufen. Weitere Informationen zu Ersatzvorlagen finden Sie unter Ersetzungsvorlagen. Weitere Informationen zur get_secret SQL Funktion finden Sie unter get_secret (secretId, geheimer Typ, Schlüssel, roleArn). Wenn der Truststore in Form einer Datei vorliegt, verwenden Sie den SecretBinary Parameter. Wenn der Truststore die Form einer Zeichenfolge hat, verwenden Sie den SecretString Parameter.

Die maximale Größe dieses Wertes beträgt 65 KB.

ssl.truststore.password

Das Passwort für den Truststore Dieser Wert ist nur erforderlich, wenn Sie ein Passwort für den Truststore erstellt haben.

ssl.keystore

Die Keystore-Datei. Dieser Wert ist erforderlich, wenn Sie SSL als Wert für security.protocol angeben.

Dieses Feld unterstützt Substitutionsvorlagen. Verwenden Sie Secrets Manager, um die Anmeldeinformationen zu speichern, die für die Verbindung mit Ihrem Kafka-Broker erforderlich sind. Verwenden Sie die get_secret SQL-Funktion, um den Wert für dieses Feld abzurufen. Weitere Informationen zu Ersatzvorlagen finden Sie unter Ersetzungsvorlagen. Weitere Informationen zur get_secret SQL Funktion finden Sie unter get_secret (secretId, geheimer Typ, Schlüssel, roleArn). Verwenden Sie den Parameter SecretBinary.

ssl.keystore.password

Das Speicherpasswort für die Keystore-Datei. Dieser Wert ist erforderlich, wenn Sie einen Wert für ssl.keystore angeben.

Der Wert dieses Feldes kann Klartext sein. Dieses Feld unterstützt auch Ersatzvorlagen. Verwenden Sie Secrets Manager, um die Anmeldeinformationen zu speichern, die für die Verbindung mit Ihrem Kafka-Broker erforderlich sind. Verwenden Sie die get_secret SQL-Funktion, um den Wert für dieses Feld abzurufen. Weitere Informationen zu Ersatzvorlagen finden Sie unter Ersetzungsvorlagen. Weitere Informationen zur get_secret SQL Funktion finden Sie unter get_secret (secretId, geheimer Typ, Schlüssel, roleArn). Verwenden Sie den Parameter SecretString.

ssl.key.password

Das Passwort des privaten Schlüssels in Ihrer Keystore-Datei.

sasl.mechanik

Der Sicherheitsmechanismus, der für die Verbindung zu Ihrem Kafka-Broker verwendet wird. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol angeben.

Zulässige Werte: PLAIN, SCRAM-SHA-512, GSSAPI.

Anmerkung

SCRAM-SHA-512ist der einzige unterstützte Sicherheitsmechanismus in den Regionen cn-north-1, cn-northwest-1, -1 und -1. us-gov-east us-gov-west

sasl.plain.username

Der Benutzername, der verwendet wird, um die geheime Zeichenfolge vom Secrets Manager abzurufen. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol und PLAIN für sasl.mechanism angeben.

sasl.plain.password

Das Passwort, das zum Abrufen der geheimen Zeichenfolge von Secrets Manager verwendet wird. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol und PLAIN für sasl.mechanism angeben.

sasl.scram.username

sasl.scram.password

Das Passwort, das zum Abrufen der geheimen Zeichenfolge von Secrets Manager verwendet wird. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol und SCRAM-SHA-512 für sasl.mechanism angeben.

sasl.kerberos.keytab

Die Keytab-Datei für die Kerberos-Authentifizierung in Secrets Manager. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol und GSSAPI für sasl.mechanism angeben.

sasl.kerberos.service.name

Der Kerberos-Prinzipalname, unter dem Apache Kafka ausgeführt wird. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol und GSSAPI für sasl.mechanism angeben.

sasl.kerberos.krb5.kdc

Der Hostname des Key Distribution Center (KDC), mit dem Ihr Apache Kafka Producer-Client eine Verbindung herstellt. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol und GSSAPI für sasl.mechanism angeben.

sasl.kerberos.krb5.realm

Der Realm, mit dem Ihr Apache Kafka Producer-Client eine Verbindung herstellt. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol und GSSAPI für sasl.mechanism angeben.

sasl.kerberos.principal

Die eindeutige Kerberos-Identität, der Kerberos Tickets für den Zugriff auf Kerberos-fähige Dienste zuweisen kann. Dieser Wert ist erforderlich, wenn Sie SASL_SSL für security.protocol und GSSAPI für sasl.mechanism angeben.

Beispiele

Das folgende JSON-Beispiel definiert eine Apache Kafka-Aktion in einer Regel. AWS IoT Im folgenden Beispiel wird die Inline-Funktion sourceIp () als Ersatzvorlage im Kafka Action-Header übergeben.


{
	"topicRulePayload": {
		"sql": "SELECT * FROM 'some/topic'",
		"ruleDisabled": false,
		"awsIotSqlVersion": "2016-03-23",
		"actions": [
			{
				"kafka": {
					"destinationArn": "arn:aws:iot:region:123456789012:ruledestination/vpc/VPCDestinationARN",
					"topic": "TopicName",
					"clientProperties": {
						"bootstrap.servers": "kafka.com:9092",
						"security.protocol": "SASL_SSL",
						"ssl.truststore": "${get_secret('kafka_client_truststore', 'SecretBinary','arn:aws:iam::123456789012:role/kafka-get-secret-role-name')}",
						"ssl.truststore.password": "kafka password",
						"sasl.mechanism": "GSSAPI",
						"sasl.kerberos.service.name": "kafka",
						"sasl.kerberos.krb5.kdc": "kerberosdns.com",
						"sasl.kerberos.keytab": "${get_secret('kafka_keytab','SecretBinary', 'arn:aws:iam::123456789012:role/kafka-get-secret-role-name')}",
						"sasl.kerberos.krb5.realm": "KERBEROSREALM",
						"sasl.kerberos.principal": "kafka-keytab/kafka-keytab.com"
					},
					"headers": [
						{
							"key": "static_header_key",
							"value": "static_header_value"
						},
						{
							"key": "substitutable_header_key",
							"value": "${value_from_payload}"
						},
						{
							"key": "source_ip",
							"value": "${sourceIp()}"
						}
					]
				}
			}
		]
	}
}

Wichtige Hinweise zu Ihrem Kerberos-Setup

Ihr Key Distribution Center (KDC) muss über ein privates Domain Name System (DNS) innerhalb Ihrer Ziel-VPC auflösbar sein. Ein möglicher Ansatz besteht darin, den KDC-DNS-Eintrag zu einer privaten gehosteten Zone hinzuzufügen. Weitere Informationen zu diesem Ansatz finden Sie unter Arbeiten mit privat gehosteten Zonen.
Für jede VPC muss die DNS-Auflösung aktiviert sein. Weitere Informationen finden Sie unter Verwenden von DNS in Ihrer VPC.
Sicherheitsgruppen für Netzwerkschnittstellen und Sicherheitsgruppen auf Instance-Ebene im VPC-Ziel müssen Datenverkehr aus Ihrer VPC an den folgenden Ports zulassen.
- TCP-Verkehr auf dem Bootstrap-Broker-Listener-Port (häufig 9092, muss aber im Bereich 9000—9100 liegen)
- TCP- und UDP-Verkehr auf Port 88 für das KDC
SCRAM-SHA-512ist der einzige unterstützte Sicherheitsmechanismus in den Regionen cn-north-1, cn-northwest-1, -1 und -1. us-gov-east us-gov-west

Warnung JavaScript ist in Ihrem Browser nicht verfügbar oder deaktiviert.

Zur Nutzung der AWS-Dokumentation muss JavaScript aktiviert sein. Weitere Informationen finden auf den Hilfe-Seiten Ihres Browsers.

Dokumentkonventionen

AWS IoT Regelaktionen

Virtual Private Cloud (VPC) -Ziele