Koneksi Kinesis - AWS Glue
Konfigurasikan KinesisBaca dari KinesisMenulis ke KinesisParameter koneksi Kinesis

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 menggunakangetSource,getSourceWithFormat, createDataFrameFromOptions ataucreate_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 dengangetSource,createDataFrameFromOptions, getSink

    • additionalOptions: Gunakan dengangetCatalogSource, getCatalogSink

    • options: Gunakan dengangetSourceWithFormat, getSinkWithFormat

  • Python

    • connection_options: Gunakan dengancreate_data_frame_from_options, write_dynamic_frame_from_options

    • additional_options: Gunakan dengancreate_data_frame_from_catalog, write_dynamic_frame_from_catalog

    • options: Gunakan dengangetSource, getSink

Untuk catatan dan batasan tentang pekerjaan Streaming ETL, konsultasikanStreaming catatan dan batasan ETL.

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.

  • 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.

Untuk informasi lebih lanjut tentang prasyarat pekerjaan Streaming ETL, lihat. Lowongan kerja Streaming ETL di AWS Glue

Contoh: Membaca dari aliran Kinesis

Contoh: Membaca dari aliran Kinesis

Digunakan bersama denganforEachBatch.

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 denganforEachBatch.

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 denganendpointUrl.

  • "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 +/-. 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 kelipatanmaxRecordPerRead. 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-turutgetRecords. 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 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 defaultnya 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 “__src_timestamp” yang menunjukkan waktu ketika catatan terkait diterima oleh aliran. Nilai defaultnya 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 catatan tertua yang diterima oleh aliran dan waktu tiba AWS Glue ke CloudWatch. Nama metriknya adalah “glue.driver.streaming. maxConsumerLagInMs”. Nilai defaultnya 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

  • "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.