Apache Kafka

Die Apache Kafka (Kafka) -Aktion sendet Nachrichten direkt an Ihr Amazon Managed Streaming for Apache Kafka (AmazonMSK), Apache Kafka-Cluster, die von Drittanbietern wie Confluent Cloud verwaltet werden, oder an selbstverwaltete Apache Kafka-Cluster. Mit Kafka-Regelaktionen können Sie Ihre IoT-Daten an Kafka-Cluster weiterleiten. Auf diese Weise können Sie leistungsstarke Datenpipelines für verschiedene Zwecke einrichten, z. B. für Streaming-Analysen, Datenintegration, Visualisierung und geschäftskritische Geschäftsanwendungen.

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. MSK Serverlose Cluster können nur über IAM Authentifizierung ausgeführt werden, was die Apache Kafka-Regelaktion derzeit nicht unterstützt. Weitere Informationen zur Konfiguration AWS IoT Core mit Confluent finden Sie unter Leveraging Confluent and Solve IoT Device and Data AWS Management Challenges.

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 Interfaces im EC2Amazon-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 zu Ihrem Kafka-Broker erforderlich sind, müssen Sie eine IAM Rolle erstellen, die die Ausführung der secretsmanager:GetSecretValue secretsmanager:DescribeSecret AND-Operationen übernehmen AWS IoT Core kann.

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 (AmazonVPC) ausführen. Sie müssen ein VPC Amazon-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 aufgelisteten Subnetze eine Netzwerkschnittstelle, über die der Datenverkehr direkt an das weitergeleitet wird. VPC Wenn Sie ein VPC Ziel erstellen, erstellt die AWS IoT Regel-Engine automatisch eine VPC Regelaktion. Weitere Informationen zu VPC Regelaktionen finden Sie unterZiele VPC für virtuelle private Clouds ().
Wenn Sie einen vom Kunden verwalteten AWS KMS key (KMSSchlüssel) verwenden, um Daten im Speicher zu verschlüsseln, muss der Dienst über die Erlaubnis verfügen, den KMS Schlüssel im Namen des Anrufers zu verwenden. Weitere Informationen finden Sie unter MSKAmazon-Verschlüsselung im Amazon Managed Streaming for Apache Kafka Developer Guide.

Parameter

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

destinationArn

Der Amazon-Ressourcenname (ARN) des VPC Ziels. Informationen zum Erstellen eines VPC Ziels finden Sie unterZiele VPC für virtuelle private Clouds ().

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 zu der get_secret SQL Funktion finden Sie unterget_secret (secretId,, SchlüsselsecretType,) 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 Hinweise zu der get_secret SQL Funktion finden Sie unterget_secret (secretId,, SchlüsselsecretType,) 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 Hinweise zu der get_secret SQL Funktion finden Sie unterget_secret (secretId,, SchlüsselsecretType,) 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.

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 Funktion, um den Wert für dieses Feld abzurufen. get_secret SQL Weitere Informationen zu Ersatzvorlagen finden Sie unter Ersetzungsvorlagen. Weitere Hinweise zu der get_secret SQL Funktion finden Sie unterget_secret (secretId,, SchlüsselsecretType,) roleArn. Verwenden Sie den Parameter SecretBinary.

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 Schlüsselverteilungszentrums (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 in den 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 Schlüsselverteilungszentrum (KDC) muss über ein privates Domain Name System (DNS) innerhalb Ihres Ziels auflösbar sein. VPC Ein möglicher Ansatz besteht darin, den KDC DNS Eintrag einer privaten gehosteten Zone hinzuzufügen. Weitere Informationen zu diesem Ansatz finden Sie unter Arbeiten mit privat gehosteten Zonen.
Für jeden VPC muss die DNS Auflösung aktiviert sein. Weitere Informationen finden Sie unter Verwenden DNS mit Ihrem VPC.
Sicherheitsgruppen für Netzwerkschnittstellen und Sicherheitsgruppen auf Instanzebene im VPC Ziel müssen den Datenverkehr innerhalb Ihrer Ports VPC an den folgenden Ports zulassen.
- TCPVerkehr auf dem Bootstrap-Broker-Listener-Port (häufig 9092, muss aber im Bereich 9000—9100 liegen)
- TCPund Verkehr auf Port 88 für UDP 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

Ziele VPC für virtuelle private Clouds ()