Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Konektor Amazon Athena Apache Kafka
Konektor Amazon Athena untuk Apache Kafka memungkinkan Amazon Athena menjalankan kueri SQL pada topik Apache Kafka Anda. Gunakan konektor ini untuk melihat topik [Apache Kafka](https://kafka.apache.org/) sebagai tabel dan pesan sebagai baris di Athena.
Konektor ini tidak menggunakan Glue Connections untuk memusatkan properti konfigurasi di Glue. Konfigurasi koneksi dilakukan melalui Lambda.
## Prasyarat
Menyebarkan konektor ke Anda Akun AWS menggunakan konsol Athena atau. AWS Serverless Application Repository Untuk informasi selengkapnya, lihat [Buat koneksi sumber data](connect-to-a-data-source.md) atau [Gunakan AWS Serverless Application Repository untuk menyebarkan konektor sumber data](connect-data-source-serverless-app-repo.md).
## Batasan
+ Menulis operasi DDL tidak didukung.
+ Batas Lambda yang relevan. Untuk informasi selengkapnya, lihat [Kuota Lambda](https://docs.aws.amazon.com/lambda/latest/dg/gettingstarted-limits.html) dalam *Panduan Pengembang AWS Lambda *.
+ Tipe data tanggal dan stempel waktu dalam kondisi filter harus dilemparkan ke tipe data yang sesuai.
+ Tipe data tanggal dan stempel waktu tidak didukung untuk jenis file CSV dan diperlakukan sebagai nilai varchar.
+ Pemetaan ke bidang JSON bersarang tidak didukung. Konektor hanya memetakan bidang tingkat atas.
+ Konektor tidak mendukung tipe yang kompleks. Tipe kompleks ditafsirkan sebagai string.
+ Untuk mengekstrak atau bekerja dengan nilai JSON yang kompleks, gunakan fungsi terkait JSON yang tersedia di Athena. Untuk informasi selengkapnya, lihat [Ekstrak data JSON dari string](extracting-data-from-JSON.md).
+ Konektor tidak mendukung akses ke metadata pesan Kafka.
## Ketentuan
+ **Metadata handler** — Penangan Lambda yang mengambil metadata dari instance database Anda.
+ **Record handler** - Handler Lambda yang mengambil catatan data dari instance database Anda.
+ **Composite handler** — Handler Lambda yang mengambil data metadata dan data dari instance database Anda.
+ **Endpoint Kafka** — String teks yang membuat koneksi ke instance Kafka.
## Kompatibilitas cluster
Konektor Kafka dapat digunakan dengan jenis cluster berikut.
+ **Standalone Kafka** — Koneksi langsung ke Kafka (diautentikasi atau tidak diautentikasi).
+ **Confluent** — Koneksi langsung ke Confluent Kafka. *Untuk informasi tentang penggunaan Athena dengan data Confluent Kafka, lihat [Memvisualisasikan data Confluent di Quick menggunakan Amazon Athena di Blog Business](https://aws.amazon.com/blogs/business-intelligence/visualize-confluent-data-in-amazon-quicksight-using-amazon-athena/) Intelligence.AWS *
### Menghubungkan ke Confluent
Menghubungkan ke Confluent membutuhkan langkah-langkah berikut:
1. Buat kunci API dari Confluent.
1. Simpan nama pengguna dan kata sandi untuk kunci Confluent API ke dalam. AWS Secrets Manager
1. Berikan nama rahasia untuk variabel `secrets_manager_secret` lingkungan di konektor Kafka.
1. Ikuti langkah-langkah di [Menyiapkan konektor Kafka](#connectors-kafka-setup) bagian dokumen ini.
## Metode otentikasi yang didukung
Konektor mendukung metode otentikasi berikut.
+ [SSL](https://kafka.apache.org/documentation/#security_ssl)
+ [SELEMPANG](https://kafka.apache.org/documentation/#security_sasl_scram)
+ SASL/PLAIN
+ SELEMPANG/PLAINTEXT
+ NO\$1AUTH
+ **Platform Kafka dan Confluent yang dikelola sendiri** - SSL,, NO\$1AUTH SASL/SCRAM, SASL/PLAINTEXT
+ **Kafka dan Confluent Cloud yang dikelola sendiri** - SASL/PLAIN
Untuk informasi selengkapnya, lihat [Mengkonfigurasi otentikasi untuk konektor Athena Kafka](#connectors-kafka-setup-configuring-authentication).
## Format data masukan yang didukung
Konektor mendukung format data input berikut.
+ JSON
+ CSV
+ AVRO
+ PROTOBUF (BUFFER PROTOKOL)
## Parameter
Gunakan parameter di bagian ini untuk mengkonfigurasi konektor Athena Kafka.
+ **auth\$1type** - Menentukan jenis otentikasi cluster. Konektor mendukung jenis otentikasi berikut:
+ **NO\$1AUTH —** Connect langsung ke Kafka (misalnya, ke cluster Kafka yang digunakan melalui instance EC2 yang tidak menggunakan otentikasi).
+ **SASL\$1SSL\$1PLAIN** — Metode ini menggunakan protokol `SASL_SSL` keamanan dan mekanisme `PLAIN` SASL. Untuk informasi selengkapnya, lihat [konfigurasi SASL](https://kafka.apache.org/documentation/#security_sasl_config) dalam dokumentasi Apache Kafka.
+ **SASL\$1PLAINTEXT\$1PLAIN** — Metode ini menggunakan protokol `SASL_PLAINTEXT` keamanan dan mekanisme `PLAIN` SASL. Untuk informasi selengkapnya, lihat [konfigurasi SASL](https://kafka.apache.org/documentation/#security_sasl_config) dalam dokumentasi Apache Kafka.
+ **SASL\$1SSL\$1SCRAM\$1 SHA512** — Anda dapat menggunakan jenis otentikasi ini untuk mengontrol akses ke cluster Apache Kafka Anda. Metode ini menyimpan nama pengguna dan kata sandi di AWS Secrets Manager. Rahasianya harus dikaitkan dengan cluster Kafka. Untuk informasi selengkapnya, lihat [Otentikasi menggunakan SASL/SCRAM](https://kafka.apache.org/documentation/#security_sasl_scram) di dokumentasi Apache Kafka.
+ **SASL\$1PLAINTEXT\$1SCRAM\$1 SHA512** — Metode ini menggunakan protokol `SASL_PLAINTEXT` keamanan dan mekanisme. `SCRAM_SHA512 SASL` Metode ini menggunakan nama pengguna dan kata sandi Anda yang disimpan di AWS Secrets Manager. Untuk informasi selengkapnya, lihat bagian [konfigurasi SASL](https://kafka.apache.org/documentation/#security_sasl_config) dari dokumentasi Apache Kafka.
+ **SSL** — Otentikasi SSL menggunakan penyimpanan kunci dan file toko kepercayaan untuk terhubung dengan cluster Apache Kafka. Anda harus membuat file trust store dan key store, mengunggahnya ke bucket Amazon S3, dan memberikan referensi ke Amazon S3 saat Anda menggunakan konektor. Toko kunci, toko kepercayaan, dan kunci SSL disimpan di dalamnya AWS Secrets Manager. Klien Anda harus memberikan kunci AWS rahasia saat konektor dikerahkan. Untuk informasi selengkapnya, lihat [Enkripsi dan Otentikasi menggunakan SSL di dokumentasi](https://kafka.apache.org/documentation/#security_ssl) Apache Kafka.
Untuk informasi selengkapnya, lihat [Mengkonfigurasi otentikasi untuk konektor Athena Kafka](#connectors-kafka-setup-configuring-authentication).
+ **certificates\$1s3\$1reference** — Lokasi Amazon S3 yang berisi sertifikat (penyimpanan kunci dan file penyimpanan kepercayaan).
+ **disable\$1spill\$1encryption** — (Opsional) Ketika diatur ke, menonaktifkan enkripsi tumpahan. `True` Defaultnya `False` sehingga data yang tumpah ke S3 dienkripsi menggunakan AES-GCM — baik menggunakan kunci yang dihasilkan secara acak atau KMS untuk menghasilkan kunci. [Menonaktifkan enkripsi tumpahan dapat meningkatkan kinerja, terutama jika lokasi tumpahan Anda menggunakan enkripsi sisi server.](https://docs.aws.amazon.com/AmazonS3/latest/userguide/serv-side-encryption.html)
+ **kafka\$1endpoint — Detail titik akhir** untuk diberikan kepada Kafka.
+ **schema\$1registry\$1url — Alamat URL** untuk registri skema (misalnya,). `http://schema-registry.example.org:8081` Berlaku untuk format `AVRO` dan `PROTOBUF` data. Athena hanya mendukung registri skema Confluent.
+ **secrets\$1manager\$1secret** — Nama rahasia tempat AWS kredensialnya disimpan.
+ **Parameter tumpahan** — Fungsi Lambda menyimpan sementara (“tumpahan”) data yang tidak sesuai dengan memori ke Amazon S3. Semua instance database yang diakses oleh fungsi Lambda yang sama tumpah ke lokasi yang sama. Gunakan parameter dalam tabel berikut untuk menentukan lokasi tumpahan.
****
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/athena/latest/ug/connectors-kafka.html)
+ **Subnet IDs** — Satu atau lebih subnet IDs yang sesuai dengan subnet yang dapat digunakan fungsi Lambda untuk mengakses sumber data Anda.
+ Cluster **Kafka Publik atau cluster Confluent Cloud standar** — Kaitkan konektor dengan subnet pribadi yang memiliki NAT Gateway.
+ **Cluster Confluent Cloud dengan konektivitas pribadi** — Kaitkan konektor dengan subnet pribadi yang memiliki rute ke cluster Confluent Cloud.
+ Untuk [AWS Transit Gateway](https://docs.confluent.io/cloud/current/networking/aws-transit-gateway.html), subnet harus berada dalam VPC yang dilampirkan ke gateway transit yang sama yang digunakan Confluent Cloud.
+ Untuk [VPC Peering](https://docs.confluent.io/cloud/current/networking/peering/aws-peering.html), subnet harus dalam VPC yang diintip ke Confluent Cloud VPC.
+ Untuk [AWS PrivateLink](https://docs.confluent.io/cloud/current/networking/private-links/aws-privatelink.html), subnet harus dalam VPC yang memiliki rute ke titik akhir VPC yang terhubung ke Confluent Cloud.
**catatan**
Jika Anda menyebarkan konektor ke VPC untuk mengakses sumber daya pribadi dan juga ingin terhubung ke layanan yang dapat diakses publik seperti Confluent, Anda harus mengaitkan konektor dengan subnet pribadi yang memiliki NAT Gateway. Untuk informasi selengkapnya, lihat [gateway NAT di Panduan](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) Pengguna Amazon VPC.
## Dukungan tipe data
Tabel berikut menunjukkan tipe data yang sesuai yang didukung untuk Kafka dan Apache Arrow.
****
| Kafka | Panah |
| --- | --- |
| CHAR | VARCHAR |
| VARCHAR | VARCHAR |
| TIMESTAMP | MILIDETIK |
| TANGGAL | DAY |
| BOOLEAN | BOOL |
| SMALLINT | SMALLINT |
| INTEGER | INT |
| BIGINT | BIGINT |
| DECIMAL | FLOAT8 |
| DOUBLE | FLOAT8 |
## Partisi dan split
Topik Kafka dibagi menjadi beberapa partisi. Setiap partisi dipesan. Setiap pesan dalam partisi memiliki ID tambahan yang disebut *offset*. Setiap partisi Kafka dibagi lagi menjadi beberapa split untuk pemrosesan paralel. Data tersedia untuk periode retensi yang dikonfigurasi dalam cluster Kafka.
## Praktik terbaik
Sebagai praktik terbaik, gunakan predikat pushdown saat Anda menanyakan Athena, seperti pada contoh berikut.
```
SELECT *
FROM "kafka_catalog_name"."glue_schema_registry_name"."glue_schema_name"
WHERE integercol = 2147483647
```
```
SELECT *
FROM "kafka_catalog_name"."glue_schema_registry_name"."glue_schema_name"
WHERE timestampcol >= TIMESTAMP '2018-03-25 07:30:58.878'
```
## Menyiapkan konektor Kafka
Sebelum Anda dapat menggunakan konektor, Anda harus mengatur cluster Apache Kafka Anda, menggunakan [AWS Glue Schema Registry](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html) untuk menentukan skema Anda, dan mengkonfigurasi otentikasi untuk konektor.
Saat bekerja dengan AWS Glue Schema Registry, perhatikan poin-poin berikut:
+ Pastikan bahwa teks di bidang **Deskripsi** dari AWS Glue Schema Registry menyertakan string`{AthenaFederationKafka}`. String penanda ini diperlukan untuk AWS Glue Registries yang Anda gunakan dengan konektor Amazon Athena Kafka.
+ Untuk kinerja terbaik, gunakan hanya huruf kecil untuk nama database dan nama tabel Anda. Menggunakan casing campuran menyebabkan konektor melakukan pencarian case insensitive yang lebih intensif secara komputasi.
**Untuk mengatur lingkungan Apache Kafka dan AWS Glue Registri Skema**
1. Siapkan lingkungan Apache Kafka Anda.
1. Unggah file deskripsi topik Kafka (yaitu skema) dalam format JSON ke Schema Registry. AWS Glue Untuk informasi selengkapnya, lihat [Mengintegrasikan dengan Registri AWS Glue Skema](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html) di Panduan AWS Glue Pengembang.
1. Untuk menggunakan format `AVRO` atau `PROTOBUF` data saat Anda menentukan skema di Registri AWS Glue Skema:
+ Untuk **nama Skema**, masukkan nama topik Kafka dalam casing yang sama dengan aslinya.
+ Untuk **format Data**, pilih **Apache Avro** atau **Protocol** Buffers.
Misalnya skema, lihat bagian berikut.
### Contoh skema untuk Registri AWS Glue Skema
Gunakan format contoh di bagian ini saat Anda mengunggah skema Anda ke Registri [AWS Glue Skema](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html).
#### Contoh skema tipe JSON
Dalam contoh berikut, skema yang akan dibuat dalam AWS Glue Schema Registry menentukan `json` sebagai nilai untuk `dataFormat` dan menggunakan untuk. `datatypejson` `topicName`
**catatan**
Nilai untuk `topicName` harus menggunakan casing yang sama dengan nama topik di Kafka.
```
{
"topicName": "datatypejson",
"message": {
"dataFormat": "json",
"fields": [
{
"name": "intcol",
"mapping": "intcol",
"type": "INTEGER"
},
{
"name": "varcharcol",
"mapping": "varcharcol",
"type": "VARCHAR"
},
{
"name": "booleancol",
"mapping": "booleancol",
"type": "BOOLEAN"
},
{
"name": "bigintcol",
"mapping": "bigintcol",
"type": "BIGINT"
},
{
"name": "doublecol",
"mapping": "doublecol",
"type": "DOUBLE"
},
{
"name": "smallintcol",
"mapping": "smallintcol",
"type": "SMALLINT"
},
{
"name": "tinyintcol",
"mapping": "tinyintcol",
"type": "TINYINT"
},
{
"name": "datecol",
"mapping": "datecol",
"type": "DATE",
"formatHint": "yyyy-MM-dd"
},
{
"name": "timestampcol",
"mapping": "timestampcol",
"type": "TIMESTAMP",
"formatHint": "yyyy-MM-dd HH:mm:ss.SSS"
}
]
}
}
```
#### Contoh skema tipe CSV
Dalam contoh berikut, skema yang akan dibuat dalam AWS Glue Schema Registry menentukan `csv` sebagai nilai untuk `dataFormat` dan menggunakan untuk. `datatypecsvbulk` `topicName` Nilai untuk `topicName` harus menggunakan casing yang sama dengan nama topik di Kafka.
```
{
"topicName": "datatypecsvbulk",
"message": {
"dataFormat": "csv",
"fields": [
{
"name": "intcol",
"type": "INTEGER",
"mapping": "0"
},
{
"name": "varcharcol",
"type": "VARCHAR",
"mapping": "1"
},
{
"name": "booleancol",
"type": "BOOLEAN",
"mapping": "2"
},
{
"name": "bigintcol",
"type": "BIGINT",
"mapping": "3"
},
{
"name": "doublecol",
"type": "DOUBLE",
"mapping": "4"
},
{
"name": "smallintcol",
"type": "SMALLINT",
"mapping": "5"
},
{
"name": "tinyintcol",
"type": "TINYINT",
"mapping": "6"
},
{
"name": "floatcol",
"type": "DOUBLE",
"mapping": "7"
}
]
}
}
```
#### Contoh skema tipe AVRO
Contoh berikut digunakan untuk membuat skema berbasis AVRO di Schema Registry. AWS Glue **Saat Anda menentukan skema di Registri AWS Glue Skema, untuk nama **Skema, Anda memasukkan nama** topik Kafka dalam casing yang sama dengan aslinya, dan untuk **format Data**, Anda memilih Apache Avro.** Karena Anda menentukan informasi ini secara langsung di registri, `topicName` bidang `dataformat` dan tidak diperlukan.
```
{
"type": "record",
"name": "avrotest",
"namespace": "example.com",
"fields": [{
"name": "id",
"type": "int"
},
{
"name": "name",
"type": "string"
}
]
}
```
#### Contoh skema tipe PROTOBUF
Contoh berikut digunakan untuk membuat skema berbasis Protobuf di Schema Registry. AWS Glue **Saat Anda menentukan skema di Registri AWS Glue Skema, untuk nama **Skema, Anda memasukkan nama** topik Kafka dalam casing yang sama dengan aslinya, dan untuk **format Data**, Anda memilih Protocol Buffer.** Karena Anda menentukan informasi ini secara langsung di registri, `topicName` bidang `dataformat` dan tidak diperlukan. Baris pertama mendefinisikan skema sebagai PROTOBUF.
```
syntax = "proto3";
message protobuftest {
string name = 1;
int64 calories = 2;
string colour = 3;
}
```
Untuk informasi selengkapnya tentang menambahkan registri dan skema di AWS Glue Schema Registry, lihat [Memulai Schema Registry di dokumentasi](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-gs.html). AWS Glue
### Mengkonfigurasi otentikasi untuk konektor Athena Kafka
Anda dapat menggunakan berbagai metode untuk mengautentikasi ke cluster Apache Kafka Anda, termasuk SSL,. SASL/SCRAM, SASL/PLAIN, and SASL/PLAINTEXT
Tabel berikut menunjukkan jenis otentikasi untuk konektor dan protokol keamanan dan mekanisme SASL untuk masing-masing. Untuk informasi lebih lanjut, lihat bagian [Keamanan](https://kafka.apache.org/documentation/#security) dari dokumentasi Apache Kafka.
****
| auth\$1type | security.protocol | sasl.mekanisme | Kompatibilitas tipe cluster |
| --- | --- | --- | --- |
| SASL\$1SSL\$1PLAIN | SASL\$1SSL | PLAIN | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/athena/latest/ug/connectors-kafka.html) |
| SASL\$1PLAINTEXT\$1PLAIN | SASL\$1PLAINTEXT | PLAIN | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/athena/latest/ug/connectors-kafka.html) |
| SASL\$1SSL\$1SCRAM\$1SHA512 | SASL\$1SSL | SCRAM-SHA-512 | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/athena/latest/ug/connectors-kafka.html) |
| SASL\$1PLAINTEXT\$1SCRAM\$1SHA512 | SASL\$1PLAINTEXT | SCRAM-SHA-512 | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/athena/latest/ug/connectors-kafka.html) |
| SSL | SSL | N/A | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/athena/latest/ug/connectors-kafka.html) |
#### SSL
Jika klaster diautentikasi SSL, Anda harus membuat file trust store dan key store dan mengunggahnya ke bucket Amazon S3. Anda harus memberikan referensi Amazon S3 ini saat Anda menggunakan konektor. Toko kunci, toko kepercayaan, dan kunci SSL disimpan di file. AWS Secrets Manager Anda memberikan kunci AWS rahasia saat Anda menggunakan konektor.
Untuk informasi tentang cara membuat rahasia di Secrets Manager, lihat [Membuat AWS Secrets Manager rahasia](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html).
Untuk menggunakan jenis otentikasi ini, atur variabel lingkungan seperti yang ditunjukkan pada tabel berikut.
****
| Parameter | Nilai |
| --- | --- |
| auth\$1type | SSL |
| certificates\$1s3\$1reference | Lokasi Amazon S3 yang berisi sertifikat. |
| secrets\$1manager\$1secret | Nama kunci AWS rahasiamu. |
Setelah Anda membuat rahasia di Secrets Manager, Anda dapat melihatnya di konsol Secrets Manager.
**Untuk melihat rahasia Anda di Secrets Manager**
1. Buka konsol Secrets Manager di [https://console.aws.amazon.com/secretsmanager/](https://console.aws.amazon.com/secretsmanager/).
1. Di panel navigasi, pilih **Rahasia**.
1. Pada halaman **Rahasia**, pilih tautan ke rahasia Anda.
1. Pada halaman detail untuk rahasia Anda, pilih **Ambil nilai rahasia**.
Gambar berikut menunjukkan contoh rahasia dengan tiga pasangan kunci/nilai:`keystore_password`,`truststore_password`, dan. `ssl_key_password`
![\[Mengambil rahasia SSL di Secrets Manager\]](http://docs.aws.amazon.com/id_id/athena/latest/ug/images/connectors-kafka-setup-1.png)
Untuk informasi selengkapnya tentang penggunaan SSL dengan Kafka, lihat [Enkripsi dan Otentikasi menggunakan SSL](https://kafka.apache.org/documentation/#security_ssl) di dokumentasi Apache Kafka.
#### SASL/SCRAM
Jika klaster Anda menggunakan otentikasi SCRAM, berikan kunci Secrets Manager yang terkait dengan cluster saat Anda menerapkan konektor. AWS Kredensi pengguna (kunci rahasia dan kunci akses) digunakan untuk mengautentikasi dengan cluster.
Mengatur variabel lingkungan seperti yang ditunjukkan pada tabel berikut.
****
| Parameter | Nilai |
| --- | --- |
| auth\$1type | SASL\$1SSL\$1SCRAM\$1SHA512 |
| secrets\$1manager\$1secret | Nama kunci AWS rahasiamu. |
Gambar berikut menunjukkan contoh rahasia di konsol Secrets Manager dengan dua key/value pasang: satu untuk`username`, dan satu untuk`password`.
![\[Mengambil rahasia SCRAM di Secrets Manager\]](http://docs.aws.amazon.com/id_id/athena/latest/ug/images/connectors-kafka-setup-2.png)
Untuk informasi selengkapnya tentang penggunaan SASL/SCRAM dengan Kafka, lihat [Otentikasi menggunakan SASL/SCRAM](https://kafka.apache.org/documentation/#security_sasl_scram) di dokumentasi Apache Kafka.
## Informasi lisensi
Dengan menggunakan konektor ini, Anda mengakui penyertaan komponen pihak ketiga, daftar yang dapat ditemukan dalam file [pom.xml](https://github.com/awslabs/aws-athena-query-federation/blob/master/athena-kafka/pom.xml) untuk konektor ini, dan menyetujui persyaratan dalam masing-masing lisensi pihak ketiga yang disediakan dalam file [LICENSE.txt](https://github.com/awslabs/aws-athena-query-federation/blob/master/athena-kafka/LICENSE.txt) di .com. GitHub
## Sumber daya tambahan
Untuk informasi tambahan tentang konektor ini, kunjungi [situs terkait](https://github.com/awslabs/aws-athena-query-federation/tree/master/athena-kafka) GitHub di.com.