Apache Kafka

Tindakan Apache Kafka (Kafka) mengirimkan pesan langsung ke Amazon Managed Streaming untuk Apache Kafka (Amazon MSK), cluster Apache Kafka yang dikelola oleh penyedia pihak ketiga seperti Confluent Cloud, atau cluster Apache Kafka yang dikelola sendiri. Dengan tindakan aturan Kafka, Anda dapat merutekan data IoT Anda ke cluster Kafka. Ini memungkinkan Anda membangun pipeline data berkinerja tinggi untuk berbagai keperluan, seperti analitik streaming, integrasi data, visualisasi, dan aplikasi bisnis penting.

catatan

Topik ini mengasumsikan keakraban dengan platform Apache Kafka dan konsep terkait. Untuk informasi lebih lanjut tentang Apache Kafka, lihat Apache Kaf ka. MSK Serverless tidak didukung. Cluster tanpa server MSK hanya dapat dilakukan melalui otentikasi IAM, yang saat ini tidak didukung tindakan aturan Apache Kafka. Untuk informasi selengkapnya tentang cara mengonfigurasi AWS IoT Core dengan Confluent, lihat Meman faatkan Confluent dan Menyelesaikan T AWS antangan Perangkat dan Manajemen Data IoT.

Persyaratan

Tindakan aturan ini memiliki persyaratan sebagai berikut:

Peran IAM yang AWS IoT dapat diasumsikan untuk melakukan ec2:DescribeSecurityGroups operasi ec2:CreateNetworkInterfaceec2:DescribeNetworkInterfaces,ec2:CreateNetworkInterfacePermission,ec2:DeleteNetworkInterface,ec2:DescribeSubnets,ec2:DescribeVpcs,ec2:DescribeVpcAttribute, dan. Peran ini membuat dan mengelola antarmuka jaringan elastis ke Amazon Virtual Private Cloud Anda untuk menjangkau broker Kafka Anda. Untuk informasi selengkapnya, lihat Pemberian AWS IoT mengatur akses yang dibutuhkannya.

Di AWS IoT konsol, Anda dapat memilih atau membuat peran AWS IoT Core untuk memungkinkan melakukan tindakan aturan ini.

Untuk informasi selengkapnya tentang antarmuka jaringan, lihat Antarmuka jaringan elastis di Panduan Pengguna Amazon EC2.

Kebijakan yang dilampirkan pada peran yang Anda tentukan harus terlihat seperti contoh berikut.
```
{
    "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": "*"
        }
    ]
}                     
```

Jika Anda menggunakan AWS Secrets Manager untuk menyimpan kredenSIAL yang diperlukan untuk terhubung ke broker Kafka Anda, Anda harus membuat peran IAM yang AWS IoT Core dapat diasumsikan untuk melakukan operasi secretsmanager:GetSecretValue dansecretsmanager:DescribeSecret.

Kebijakan yang dilampirkan pada peran yang Anda tentukan harus terlihat seperti contoh berikut.


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

Anda dapat menjalankan cluster Apache Kafka Anda di dalam Amazon Virtual Private Cloud (Amazon VPC). Anda harus membuat tujuan Apache Kafka Virtual Private Cloud (VPC) dan menggunakan gateway NAT di subnet Anda untuk meneruskan pesan dari AWS IoT klaster Kafka publik. Mesin AWS IoT aturan membuat antarmuka jaringan di setiap subnet yang terdaftar di tujuan untuk merutekan lalu lintas langsung ke VPC. Saat tujuan Anda, mesin AWS IoT aturan secara otomatis membuat tindakan aturan VPC. Untuk informasi selengkapnya tentang tindakan aturan VPC, lihatTujuan Cloud Privat Virtual (VPC) Apache Kafka.
Jika Anda menggunakan kunci yang dikelola pelanggan AWS KMS key (kunci KMS) untuk mengenkripsi data saat diam, layanan harus memiliki izin untuk menggunakan kunci KMS atas nama penelepon. Untuk informasi selengkapnya, lihat enkripsi Amazon MSK di Panduan Pengembang Amazon Managed Streaming untuk Apache Kafka.

Parameter

Saat Anda membuat AWS IoT aturan dengan tindakan ini, Anda harus menentukan informasi berikut:

DestinasiArn

Nama Sumber Daya Amazon (ARN) dari tujuan Apache Kafka Virtual Private Cloud (VPC). Untuk informasi tentang membuat tujuan, lihatTujuan Cloud Privat Virtual (VPC) Apache Kafka.

topik

Topik Kafka untuk pesan yang akan dikirim ke broker Kafka.

Anda dapat mengganti bidang ini menggunakan template substitusi. Untuk informasi selengkapnya, lihat Templat substitusi.

kunci (opsional)

Kunci pesan Kafka.

Anda dapat mengganti bidang ini menggunakan template substitusi. Untuk informasi selengkapnya, lihat Templat substitusi.

header (opsional)

Daftar header Kafka yang Anda tentukan. Setiap header adalah pasangan kunci-nilai yang dapat Anda tentukan saat Anda membuat tindakan Kafka. Anda dapat menggunakan header ini untuk merutekan data dari klien IoT ke cluster Kafka hilir tanpa memodifikasi payload pesan Anda.

Anda dapat mengganti bidang ini menggunakan template substitusi. Untuk memahami cara meneruskan fungsi Aturan sebaris sebagai template substitusi di header Tindakan Kafka, lihat Contoh. Untuk informasi selengkapnya, lihat Templat substitusi.

catatan

Header dalam format biner tidak didukung.

partisi (opsional)

Partisi pesan Kafka.

Anda dapat mengganti bidang ini menggunakan template substitusi. Untuk informasi selengkapnya, lihat Templat substitusi.

Properti Klien

Objek yang mendefinisikan properti klien produsen Apache Kafka.

acks (opsional)

Jumlah ucapan terima kasih yang diminta produsen untuk menerima server sebelum mempertimbangkan permintaan selesai.

Jika Anda menentukan 0 sebagai nilai, produsen tidak akan menunggu pengakuan dari server. Jika server tidak menerima pesan, produsen tidak akan mencoba lagi untuk mengirim pesan.

Nilai yang valid:-1,0,1,all. Nilai default-nya adalah 1.

bootstrap.servers

Daftar pasangan host dan port (misalnya,host1:port1,host2:port2) yang digunakan untuk membuat koneksi awal ke cluster Kafka Anda.

tipe kompresi (opsional)

Jenis kompresi untuk semua data yang dihasilkan oleh produsen.

Nilai yang valid:none,gzip,snappy,lz4,zstd. Nilai default-nya adalah none.

security.protocol

Protokol keamanan yang digunakan untuk melampirkan ke broker Kafka Anda.

Nilai-nilai yang valid: SSL, SASL_SSL. Nilai default-nya adalah SSL.

keyserializer

Menentukan cara mengubah objek kunci yang Anda berikan ProducerRecord menjadi byte.

Nilai valid: StringSerializer.

nilai.serializer

Menentukan cara mengubah objek nilai yang Anda berikan ProducerRecord menjadi byte.

Nilai valid: ByteBufferSerializer.

ssl.truststore

File truststore dalam format base64 atau lokasi file truststore di. AWS Secrets Manager Nilai ini tidak diperlukan jika truststore Anda dipercaya oleh otoritas sertifikat Amazon (CA).

Bidang ini mendukung template substitusi. Jika Anda menggunakan Secrets Manager untuk menyimpan kredenSIAL yang diperlukan untuk terhubung ke broker Kafka Anda, Anda dapat menggunakan fungsi get_secret SQL untuk mengambil nilai untuk bidang ini. Untuk informasi selengkapnya tentang template substitusi, lihatTemplat substitusi. Untuk informasi selengkapnya tentang fungsi get_secret SQL, lihatget_secret (SecretID, SecretType, kunci, roLearn). Jika truststore dalam bentuk file, gunakan SecretBinary parameter. Jika truststore dalam bentuk string, gunakan SecretString parameter.

Ukuran maksimum nilai ini adalah 65 KB.

ssl.truststore.password

Kata sandi untuk truststore. Nilai ini hanya diperlukan jika Anda telah membuat kata sandi untuk truststore.

ssl.keystore

File keystore. Nilai ini diperlukan saat Anda menentukan SSL sebagai nilai untuksecurity.protocol.

Bidang ini mendukung template substitusi. Gunakan Secrets Manager untuk menyimpan kredenSIAL yang diperlukan untuk terhubung ke broker Kafka Anda. Untuk mengambil nilai untuk bidang ini, gunakan fungsi get_secret SQL. Untuk informasi selengkapnya tentang template substitusi, lihatTemplat substitusi. Untuk informasi selengkapnya tentang fungsi get_secret SQL, lihatget_secret (SecretID, SecretType, kunci, roLearn). Gunakan parameter SecretBinary.

ssl.keystore.kata sandi

Kata sandi penyimpanan untuk file keystore. Nilai ini diperlukan jika Anda menentukan nilai untukssl.keystore.

Nilai bidang ini dapat berupa plaintext. Bidang ini juga mendukung template substitusi. Gunakan Secrets Manager untuk menyimpan kredenSIAL yang diperlukan untuk terhubung ke broker Kafka Anda. Untuk mengambil nilai untuk bidang ini, gunakan fungsi get_secret SQL. Untuk informasi selengkapnya tentang template substitusi, lihatTemplat substitusi. Untuk informasi selengkapnya tentang fungsi get_secret SQL, lihatget_secret (SecretID, SecretType, kunci, roLearn). Gunakan parameter SecretString.

ssl.key.kata sandi

Kata sandi kunci pribadi di file keystore Anda.

sasl.mekanisme

Mekanisme keamanan yang digunakan untuk terhubung ke broker Kafka Anda. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuksecurity.protocol.

Nilai-nilai yang valid: PLAIN, SCRAM-SHA-512, GSSAPI.

catatan

SCRAM-SHA-512adalah satu-satunya mekanisme keamanan yang didukung di Wilayah cn-nord-1, cn-northwest-1, us-gov-east-1, dan us-gov-west-1.

sasl.plain.nama pengguna

Nama pengguna yang digunakan untuk mengambil string rahasia dari Secrets Manager. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan PLAIN untuksasl.mechanism.

sasl.plain.password

Kata sandi yang digunakan untuk mengambil string rahasia dari Secrets Manager. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan PLAIN untuksasl.mechanism.

sasl.scram.nama pengguna

Nama pengguna yang digunakan untuk mengambil string rahasia dari Secrets Manager. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan SCRAM-SHA-512 untuksasl.mechanism.

sasl.scram.password

Kata sandi yang digunakan untuk mengambil string rahasia dari Secrets Manager. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan SCRAM-SHA-512 untuksasl.mechanism.

sasl.kerberos.keytab

File keytab untuk otentikasi Kerberos di Secrets Manager. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan GSSAPI untuksasl.mechanism.

sasl.kerberos.service.name

Nama utama Kerberos di mana Apache Kafka berjalan. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan GSSAPI untuksasl.mechanism.

sasl.kerberos.krb5.kdc

Nama host pusat distribusi kunci (KDC) tempat klien produsen Apache Kafka Anda terhubung. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan GSSAPI untuksasl.mechanism.

sasl.kerberos.krb5.alam

Wilayah tempat klien produsen Apache Kafka Anda terhubung. Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan GSSAPI untuksasl.mechanism.

sasl.kerberos.utama

Identitas Kerberos unik di mana Kerberos dapat menetapkan tiket untuk mengakses layanan. Kerberos-aware Nilai ini diperlukan saat Anda menentukan SASL_SSL untuk security.protocol dan GSSAPI untuksasl.mechanism.

Contoh

Contoh JSON berikut mendefinisikan tindakan Apache Kafka dalam AWS IoT aturan. Contoh berikut meneruskan fungsi inline sourceIP () sebagai template substitusi di header Tindakan Kafka.


{
	"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()}"
						}
					]
				}
			}
		]
	}
}

Catatan penting tentang pengaturan Kerberos Anda

Pusat distribusi kunci (KDC) Anda harus dapat diselesaikan melalui Sistem Nama Domain pribadi (DNS) dalam VPC target Anda. Salah satu pendekatan yang mungkin adalah menambahkan entri DNS KDC ke zona pribadi yang dihosting. Untuk informasi selengkapnya tentang pendekatan ini, lihat Bek erja dengan zona pribadi yang dihosting.
Setiap VPC harus mengaktifkan resolusi DNS. Untuk informasi selengkapnya, lihat Menggunakan DNS dengan VPC Anda.
Grup keamanan antarmuka jaringan dan grup keamanan tingkat instance di tujuan VPC harus mengizinkan lalu lintas dari dalam VPC Anda pada port berikut.
- Lalu lintas TCP pada port pendengar broker bootstrap (seringkali 9092, tetapi harus dalam kisaran 9000—9100)
- Lalu lintas TCP dan UDP pada port 88 untuk KDC
SCRAM-SHA-512adalah satu-satunya mekanisme keamanan yang didukung di Wilayah cn-nord-1, cn-northwest-1, us-gov-east-1, dan us-gov-west-1.

Awas Javascript dinonaktifkan atau tidak tersedia di browser Anda.

Untuk menggunakan Dokumentasi AWS, Javascript harus diaktifkan. Lihat halaman Bantuan browser Anda untuk petunjuk.

Konvensi Dokumen

AWS IoT tindakan aturan

Tujuan Cloud Privat Virtual (VPC) Apache Kafka