Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Koneksi Kinesis
Anda dapat menggunakan koneksi Kinesis untuk membaca dan menulis ke aliran data Amazon Kinesis menggunakan informasi yang disimpan dalam tabel Katalog Data, atau dengan memberikan informasi untuk langsung mengakses aliran data. Anda dapat membaca informasi dari Kinesis menjadi Spark DataFrame, lalu mengubahnya menjadi Glue. AWS DynamicFrame Anda dapat menulis DynamicFrames ke Kinesis dalam format JSON. Jika Anda langsung mengakses aliran data, gunakan opsi ini untuk memberikan informasi tentang cara mengakses aliran data.
Jika Anda menggunakan `getCatalogSource` atau `create_data_frame_from_catalog` menggunakan catatan dari sumber streaming Kinesis, pekerjaan tersebut memiliki database Katalog Data dan informasi nama tabel, dan dapat menggunakannya untuk mendapatkan beberapa parameter dasar untuk membaca dari sumber streaming Kinesis. Jika Anda menggunakan`getSource`,`getSourceWithFormat`, `createDataFrameFromOptions` atau`create_data_frame_from_options`, Anda harus menentukan parameter dasar ini menggunakan opsi koneksi yang dijelaskan di sini.
Anda dapat menentukan opsi koneksi untuk Kinesis menggunakan argumen berikut untuk metode yang ditentukan di `GlueContext` kelas.
+ Skala
+ `connectionOptions`: Gunakan dengan`getSource`,`createDataFrameFromOptions`, `getSink`
+ `additionalOptions`: Gunakan dengan`getCatalogSource`, `getCatalogSink`
+ `options`: Gunakan dengan`getSourceWithFormat`, `getSinkWithFormat`
+ Python
+ `connection_options`: Gunakan dengan`create_data_frame_from_options`, `write_dynamic_frame_from_options`
+ `additional_options`: Gunakan dengan`create_data_frame_from_catalog`, `write_dynamic_frame_from_catalog`
+ `options`: Gunakan dengan`getSource`, `getSink`
Untuk catatan dan batasan tentang pekerjaan Streaming ETL, konsultasikan[Streaming catatan dan batasan ETL](add-job-streaming.md#create-job-streaming-restrictions).
## Konfigurasikan Kinesis
Untuk terhubung ke aliran data Kinesis dalam pekerjaan AWS Glue Spark, Anda memerlukan beberapa prasyarat:
+ Jika membaca, tugas AWS Glue harus memiliki izin IAM tingkat akses Baca ke aliran data Kinesis.
+ Jika menulis, pekerjaan AWS Glue harus memiliki izin IAM tingkat akses Tulis ke aliran data Kinesis.
Dalam kasus tertentu, Anda perlu mengkonfigurasi prasyarat tambahan:
+ Jika pekerjaan AWS Glue Anda dikonfigurasi dengan **koneksi jaringan Tambahan** (biasanya untuk terhubung ke kumpulan data lain) dan salah satu koneksi tersebut menyediakan **opsi Jaringan** VPC Amazon, ini akan mengarahkan pekerjaan Anda untuk berkomunikasi melalui Amazon VPC. Dalam hal ini Anda juga perlu mengonfigurasi aliran data Kinesis Anda untuk berkomunikasi melalui Amazon VPC. Anda dapat melakukan ini dengan membuat titik akhir VPC antarmuka antara VPC Amazon dan aliran data Kinesis Anda. Untuk informasi selengkapnya, lihat [Menggunakan Kinesis Data Streams dengan Titik Akhir VPC Antarmuka](https://docs.aws.amazon.com//streams/latest/dev/vpc.html).
+ Saat menentukan Amazon Kinesis Data Streams di akun lain, Anda harus menyiapkan peran dan kebijakan untuk mengizinkan akses lintas akun. Untuk informasi selengkapnya, lihat [ Contoh: Baca Dari Pengaliran Kinesis di Akun Berbeda](https://docs.aws.amazon.com/kinesisanalytics/latest/java/examples-cross.html).
Untuk informasi lebih lanjut tentang prasyarat pekerjaan Streaming ETL, lihat. [Lowongan kerja Streaming ETL di AWS Glue](add-job-streaming.md)
## Contoh: Membaca dari aliran Kinesis
### Contoh: Membaca dari aliran Kinesis
Digunakan bersama dengan[forEachBatch](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-forEachBatch).
Contoh untuk sumber streaming Amazon Kinesis:
```
kinesis_options =
{ "streamARN": "arn:aws:kinesis:us-east-2:777788889999:stream/fromOptionsStream",
"startingPosition": "TRIM_HORIZON",
"inferSchema": "true",
"classification": "json"
}
data_frame_datasource0 = glueContext.create_data_frame.from_options(connection_type="kinesis", connection_options=kinesis_options)
```
## Contoh: Menulis ke aliran Kinesis
### Contoh: Membaca dari aliran Kinesis
Digunakan bersama dengan[forEachBatch](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-forEachBatch).
Contoh untuk sumber streaming Amazon Kinesis:
```
kinesis_options =
{ "streamARN": "arn:aws:kinesis:us-east-2:777788889999:stream/fromOptionsStream",
"startingPosition": "TRIM_HORIZON",
"inferSchema": "true",
"classification": "json"
}
data_frame_datasource0 = glueContext.create_data_frame.from_options(connection_type="kinesis", connection_options=kinesis_options)
```
## Referensi opsi koneksi Kinesis
Menetapkan opsi koneksi untuk Amazon Kinesis Data Streams.
Gunakan opsi koneksi berikut untuk sumber data streaming Kinesis:
+ `"streamARN"`(Wajib) Digunakan untuk Baca/Tulis. ARN dari aliran data Kinesis.
+ `"classification"`(Diperlukan untuk dibaca) Digunakan untuk Baca. Format file yang digunakan oleh data dalam catatan. Diperlukan kecuali disediakan melalui Katalog Data.
+ `"streamName"`— (Opsional) Digunakan untuk Baca. Nama aliran data Kinesis untuk dibaca. Digunakan dengan`endpointUrl`.
+ `"endpointUrl"`— (Opsional) Digunakan untuk Baca. Default: "https://kinesis.us-east-1.amazonaws.com”. AWS Titik akhir dari aliran Kinesis. Anda tidak perlu mengubah ini kecuali Anda terhubung ke wilayah khusus.
+ `"partitionKey"`— (Opsional) Digunakan untuk Menulis. Kunci partisi Kinesis digunakan saat memproduksi catatan.
+ `"delimiter"`(Opsional) Digunakan untuk Baca. Pemisah nilai yang digunakan saat `classification` CSV. Defaultnya adalah "`,`.”
+ `"startingPosition"`: (Opsional) Digunakan untuk Baca. Posisi awal dalam aliran data Kinesis untuk membaca data dari. Nilai yang mungkin adalah`"latest"`,, `"trim_horizon"``"earliest"`, atau string Timestamp dalam format UTC dalam pola `yyyy-mm-ddTHH:MM:SSZ` (di mana `Z` mewakili offset zona waktu UTC dengan \$1/-. Misalnya “2023-04-04T 08:00:00-04:00 “). Nilai default-nya adalah `"latest"`. Catatan: string Timestamp dalam Format UTC untuk hanya didukung untuk `"startingPosition"` AWS Glue versi 4.0 atau yang lebih baru.
+ `"failOnDataLoss"`: (Opsional) Gagal pekerjaan jika ada pecahan aktif yang hilang atau kedaluwarsa. Nilai default-nya adalah `"false"`.
+ `"awsSTSRoleARN"`: (Opsional) Digunakan untuk Baca/Tulis. Nama Sumber Daya Amazon (ARN) dari peran yang akan diasumsikan menggunakan AWS Security Token Service ()AWS STS. Peran ini harus memiliki izin untuk mendeskripsikan atau membaca operasi rekaman untuk aliran data Kinesis. Anda harus menggunakan parameter ini saat mengakses aliran data di akun yang berbeda. Digunakan bersama dengan`"awsSTSSessionName"`.
+ `"awsSTSSessionName"`: (Opsional) Digunakan untuk Baca/Tulis. Pengidentifikasi untuk sesi dengan asumsi peran menggunakan. AWS STS Anda harus menggunakan parameter ini saat mengakses aliran data di akun yang berbeda. Digunakan bersama dengan`"awsSTSRoleARN"`.
+ `"awsSTSEndpoint"`: (Opsional) AWS STS Titik akhir yang digunakan saat menghubungkan ke Kinesis dengan peran yang diasumsikan. Ini memungkinkan penggunaan AWS STS titik akhir regional dalam VPC, yang tidak dimungkinkan dengan titik akhir global default.
+ `"maxFetchTimeInMs"`: (Opsional) Digunakan untuk Baca. Waktu maksimum yang dihabiskan untuk pelaksana pekerjaan untuk membaca catatan untuk batch saat ini dari aliran data Kinesis, ditentukan dalam milidetik (ms). Beberapa panggilan `GetRecords` API dapat dilakukan dalam waktu ini. Nilai default-nya adalah `1000`.
+ `"maxFetchRecordsPerShard"`: (Opsional) Digunakan untuk Baca. Jumlah maksimum catatan yang diambil per pecahan dalam aliran data Kinesis per mikrobatch. Catatan: Klien dapat melampaui batas ini jika pekerjaan streaming telah membaca catatan tambahan dari Kinesis (dalam panggilan get-records yang sama). Jika `maxFetchRecordsPerShard` perlu ketat maka itu harus kelipatan`maxRecordPerRead`. Nilai default-nya adalah `100000`.
+ `"maxRecordPerRead"`: (Opsional) Digunakan untuk Baca. Jumlah maksimum catatan untuk diambil dari aliran data Kinesis dalam `getRecords` setiap operasi. Nilai default-nya adalah `10000`.
+ `"addIdleTimeBetweenReads"`: (Opsional) Digunakan untuk Baca. Menambahkan penundaan waktu antara dua operasi berturut-turut`getRecords`. Nilai default-nya adalah `"False"`. Opsi ini hanya dapat dikonfigurasi untuk Glue versi 2.0 dan di atasnya.
+ `"idleTimeBetweenReadsInMs"`: (Opsional) Digunakan untuk Baca. Waktu tunda minimum antara dua `getRecords` operasi berturut-turut, ditentukan dalam ms. Nilai default-nya adalah `1000`. Opsi ini hanya dapat dikonfigurasi untuk Glue versi 2.0 dan di atasnya.
+ `"describeShardInterval"`: (Opsional) Digunakan untuk Baca. Interval waktu minimum antara dua panggilan `ListShards` API untuk skrip Anda untuk mempertimbangkan resharding. Untuk informasi selengkapnya, lihat [Strategi Penyerpihan Kembali](https://docs.aws.amazon.com//streams/latest/dev/kinesis-using-sdk-java-resharding-strategies.html) di *Panduan Developer Amazon Kinesis Data Streams*. Nilai default-nya adalah `1s`.
+ `"numRetries"`: (Opsional) Digunakan untuk Baca. Jumlah maksimum percobaan ulang untuk permintaan API Kinesis Data Streams. Nilai default-nya adalah `3`.
+ `"retryIntervalMs"`: (Opsional) Digunakan untuk Baca. Periode waktu pendinginan (ditentukan dalam ms) sebelum mencoba kembali panggilan API Kinesis Data Streams. Nilai default-nya adalah `1000`.
+ `"maxRetryIntervalMs"`: (Opsional) Digunakan untuk Baca. Periode waktu pendinginan maksimum (ditentukan dalam ms) antara dua percobaan ulang panggilan API Kinesis Data Streams. Nilai default-nya adalah `10000`.
+ `"avoidEmptyBatches"`: (Opsional) Digunakan untuk Baca. Hindari membuat pekerjaan microbatch kosong dengan memeriksa data yang belum dibaca di aliran data Kinesis sebelum batch dimulai. Nilai default-nya adalah `"False"`.
+ `"schema"`: (Diperlukan saat InferSchema disetel ke false) Digunakan untuk Baca. Skema yang digunakan untuk memproses muatan. Jika klasifikasi adalah `avro` skema yang disediakan harus dalam format skema Avro. Jika klasifikasi tidak, skema `avro` yang disediakan harus dalam format skema DDL.
Berikut ini adalah contoh skema.
------
#### [ Example in DDL schema format ]
```
`column1` INT, `column2` STRING , `column3` FLOAT
```
------
#### [ Example in Avro schema format ]
```
{
"type":"array",
"items":
{
"type":"record",
"name":"test",
"fields":
[
{
"name":"_id",
"type":"string"
},
{
"name":"index",
"type":
[
"int",
"string",
"float"
]
}
]
}
}
```
------
+ `"inferSchema"`: (Opsional) Digunakan untuk Baca. Nilai default adalah 'salah'. Jika disetel ke 'true', skema akan terdeteksi saat runtime dari payload di dalamnya. `foreachbatch`
+ `"avroSchema"`: (Usang) Digunakan untuk Baca. Parameter yang digunakan untuk menentukan skema data Avro saat format Avro digunakan. Parameter ini sekarang tidak digunakan lagi. Gunakan parameter `schema`.
+ `"addRecordTimestamp"`: (Opsional) Digunakan untuk Baca. Ketika opsi ini diatur ke 'true', output data akan berisi kolom tambahan bernama “\$1\$1src\$1timestamp” yang menunjukkan waktu ketika catatan terkait diterima oleh aliran. Nilai default adalah 'salah'. Opsi ini didukung di AWS Glue versi 4.0 atau yang lebih baru.
+ `"emitConsumerLagMetrics"`: (Opsional) Digunakan untuk Baca. Ketika opsi disetel ke 'true', untuk setiap batch, itu akan memancarkan metrik untuk durasi antara rekaman tertua yang diterima oleh aliran dan waktu tiba. AWS Glue CloudWatch Nama metriknya adalah “glue.driver.streaming. maxConsumerLagInMs”. Nilai default adalah 'salah'. Opsi ini didukung di AWS Glue versi 4.0 atau yang lebih baru.
+ `"fanoutConsumerARN"`: (Opsional) Digunakan untuk Baca. ARN dari konsumen aliran Kinesis untuk aliran yang ditentukan dalam. `streamARN` Digunakan untuk mengaktifkan mode fan-out yang disempurnakan untuk koneksi Kinesis Anda. Untuk informasi lebih lanjut tentang mengonsumsi aliran Kinesis dengan fan-out yang ditingkatkan, lihat. [Menggunakan fan-out yang disempurnakan dalam pekerjaan streaming Kinesis](aws-glue-programming-etl-connect-kinesis-efo.md)
+ `"recordMaxBufferedTime"`— (Opsional) Digunakan untuk Menulis. Default: 1000 (ms). Waktu maksimum sebuah rekaman di-buffer sambil menunggu untuk ditulis.
+ `"aggregationEnabled"`— (Opsional) Digunakan untuk Menulis. Default: benar. Menentukan apakah catatan harus dikumpulkan sebelum mengirim mereka ke Kinesis.
+ `"aggregationMaxSize"`— (Opsional) Digunakan untuk Menulis. Default: 51200 (byte). Jika catatan lebih besar dari batas ini, itu akan melewati agregator. Catatan Kinesis memberlakukan batas 50KB pada ukuran rekaman. Jika Anda menetapkan ini di luar 50KB, catatan kebesaran akan ditolak oleh Kinesis.
+ `"aggregationMaxCount"`— (Opsional) Digunakan untuk Menulis. Standar: 4294967295. Jumlah maksimum item untuk dikemas ke dalam catatan agregat.
+ `"producerRateLimit"`— (Opsional) Digunakan untuk Menulis. Default: 150 (%). Membatasi throughput per shard yang dikirim dari satu produsen (seperti pekerjaan Anda), sebagai persentase dari batas backend.
+ `"collectionMaxCount"`— (Opsional) Digunakan untuk Menulis. Default: 500. Jumlah maksimum item untuk dikemas ke dalam PutRecords permintaan.
+ `"collectionMaxSize"`— (Opsional) Digunakan untuk Menulis. Default: 5242880 (byte). Jumlah maksimum data untuk dikirim dengan PutRecords permintaan.
# Menggunakan fan-out yang disempurnakan dalam pekerjaan streaming Kinesis
Konsumen fan-out yang ditingkatkan dapat menerima catatan dari aliran Kinesis dengan throughput khusus yang bisa lebih besar daripada konsumen biasa. Ini dilakukan dengan mengoptimalkan protokol transfer yang digunakan untuk memberikan data kepada konsumen Kinesis, seperti pekerjaan Anda. [Untuk informasi lebih lanjut tentang Kinesis Enhanced Fan-Out, lihat dokumentasi Kinesis.](https://docs.aws.amazon.com//streams/latest/dev/enhanced-consumers.html)
Dalam mode fan-out yang disempurnakan, opsi `maxRecordPerRead` dan `idleTimeBetweenReadsInMs` koneksi tidak lagi berlaku, karena parameter tersebut tidak dapat dikonfigurasi saat menggunakan fan-out yang disempurnakan. Opsi konfigurasi untuk percobaan ulang berfungsi seperti yang dijelaskan.
Gunakan prosedur berikut untuk mengaktifkan dan menonaktifkan fan-out yang disempurnakan untuk pekerjaan streaming Anda. Anda harus mendaftarkan pengguna streaming untuk setiap pekerjaan yang akan menggunakan data dari aliran Anda.
**Untuk mengaktifkan peningkatan konsumsi fan-out pada pekerjaan Anda:**
1. Daftarkan konsumen streaming untuk pekerjaan Anda menggunakan Kinesis API. [Ikuti petunjuk untuk *mendaftarkan konsumen dengan fan-out yang disempurnakan menggunakan Kinesis Data Streams API dalam dokumentasi Kinesis*.](https://docs.aws.amazon.com//streams/latest/dev/building-enhanced-consumers-api) Anda hanya perlu mengikuti langkah pertama - menelepon [RegisterStreamConsumer](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_RegisterStreamConsumer.html). Permintaan Anda harus mengembalikan ARN,. *consumerARN*
1. Setel opsi koneksi `fanoutConsumerARN` ke *consumerARN* dalam argumen metode koneksi Anda.
1. Mulai ulang pekerjaan Anda.
**Untuk menonaktifkan peningkatan konsumsi fan-out pada pekerjaan Anda:**
1. Hapus opsi `fanoutConsumerARN` koneksi dari panggilan metode Anda.
1. Mulai ulang pekerjaan Anda.
1. [Ikuti instruksi untuk *membatalkan pendaftaran konsumen dalam dokumentasi* Kinesis.](https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-console.html) Instruksi ini berlaku untuk konsol, tetapi juga dapat dicapai melalui Kinesis API. Untuk informasi lebih lanjut tentang deregistrasi konsumen streaming melalui Kinesis API, lihat di dokumentasi Kinesis. [DeregisterStreamConsumer](https://docs.aws.amazon.com//kinesis/latest/APIReference/API_DeregisterStreamConsumer.html)