Realtime Compute for Apache Flink:Log Service (SLS)

Sintaksis

CREATE TABLE sls_table(
  a INT,
  b INT,
  c VARCHAR
) WITH (
  'connector' = 'sls',
  'endPoint' = '<yourEndPoint>',
  'project' = '<yourProjectName>',
  'logStore' = '<yourLogStoreName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}'
);

Opsi

• Umum

  Parameter	Deskripsi	Jenis	Wajib	Default	Keterangan
  connector	Konektor yang digunakan.	String	Ya	None	Atur ke sls.
  endPoint	Titik akhir Log Service (SLS).	String	Ya	None	Tentukan alamat akses VPC Log Service (SLS). Untuk informasi lebih lanjut, lihat Service endpoints. Catatan Secara default, Realtime Compute for Apache Flink tidak dapat mengakses internet. Untuk mengaktifkan akses internet dari Virtual Private Cloud (VPC) Anda, gunakan NAT Gateway. Untuk informasi lebih lanjut, lihat How do I access the Internet?. Kami menyarankan agar Anda tidak mengakses SLS melalui internet. Jika harus melakukannya, gunakan HTTPS dan aktifkan transfer acceleration. Untuk informasi lebih lanjut, lihat Manage transfer acceleration.
  project	Nama project SLS.	String	Ya	None	N/A
  logStore	Nama Logstore atau MetricStore.	String	Ya	None	Data dalam Logstore dikonsumsi dengan cara yang sama seperti data dalam MetricStore.
  accessId	ID AccessKey Akun Alibaba Cloud Anda.	String	Ya	None	Untuk informasi lebih lanjut, lihat Obtain an AccessKey pair. Penting Untuk mencegah Pasangan Kunci Akses Anda terpapar, kami menyarankan agar Anda menggunakan variabel untuk menentukan ID AccessKey dan rahasia AccessKey. Untuk informasi lebih lanjut, lihat Project variables.
  accessKey	Rahasia AccessKey Akun Alibaba Cloud Anda.	String	Ya	None

• Khusus sumber

  Parameter	Deskripsi	Tipe	Wajib	Default	Keterangan
  enableNewSource	Menentukan apakah akan menggunakan sumber data baru yang mengimplementasikan antarmuka FLIP-27.	Boolean	Tidak	false	Sumber baru dapat secara otomatis menyesuaikan perubahan shard dan mendistribusikan shard semerata mungkin di seluruh subtask sumber. Penting Opsi ini hanya didukung di VVR 8.0.9 dan versi yang lebih baru. Nilai default adalah true untuk VVR 11.1 dan versi yang lebih baru. Jika Anda mengubah nilai opsi ini, Anda tidak dapat memulihkan pekerjaan dari state yang disimpan. Untuk melanjutkan konsumsi dari offset historis, pertama-tama mulai pekerjaan dengan opsi consumerGroup untuk mencatat progres konsumsi dalam kelompok konsumen SLS. Kemudian, atur opsi consumeFromCheckpoint ke true dan restart pekerjaan tanpa state. Jika Logstore memiliki shard read-only, beberapa subtask mungkin terus meminta data dari shard lain setelah menyelesaikan tugasnya sendiri. Hal ini dapat menyebabkan workload tidak seimbang, memengaruhi performa. Untuk mengatasi masalah ini, Anda dapat menyesuaikan parallelism, mengoptimalkan strategi penjadwalan, atau menggabungkan shard kecil untuk mengurangi jumlah shard dan menyederhanakan alokasi tugas.
  shardDiscoveryIntervalMs	Interval untuk penemuan shard dinamis.	Long	Tidak	60000	Atur opsi ini ke nilai negatif untuk menonaktifkan penemuan dinamis. Satuan: milidetik. Catatan Nilai harus lebih besar dari atau sama dengan 60.000 milidetik (1 menit). Opsi ini hanya berlaku ketika enableNewSource diatur ke true. Opsi ini hanya didukung di VVR 8.0.9 dan versi yang lebih baru.
  startupMode	Mode startup untuk tabel sumber.	String	Tidak	timestamp	timestamp (default): Mengonsumsi log mulai dari waktu mulai yang ditentukan. latest: Mulai mengonsumsi log dari offset terbaru. earliest: Mulai mengonsumsi log dari offset paling awal. consumer_group: Mulai konsumsi log dari offset yang dicatat oleh kelompok konsumen. Jika kelompok konsumen belum mencatat offset konsumsi untuk suatu shard, konsumsi dimulai dari offset paling awal. Penting Untuk versi VVR sebelum 11.1, nilai consumer_group tidak didukung. Anda harus mengatur consumeFromCheckpoint ke true. Dalam kasus ini, konsumsi log dimulai dari offset yang dicatat oleh kelompok konsumen yang ditentukan, dan pengaturan mode startup tidak berlaku.
  startTime	Waktu mulai untuk konsumsi log.	String	Tidak	Waktu saat ini	Formatnya adalah yyyy-MM-dd hh:mm:ss. Opsi ini hanya berlaku ketika startupMode diatur ke timestamp. Catatan Opsi startTime dan stopTime didasarkan pada atribut __receive_time__ di SLS, bukan atribut __timestamp__.
  stopTime	Waktu akhir untuk konsumsi log.	String	Tidak	None	Formatnya adalah yyyy-MM-dd hh:mm:ss. Catatan Opsi ini hanya digunakan untuk mengonsumsi log historis dan harus diatur ke waktu di masa lalu. Jika Anda mengaturnya ke waktu di masa depan, konsumsi mungkin berhenti lebih awal jika tidak ada log baru yang ditulis, sehingga mengakibatkan gangguan aliran data tanpa pesan error. Jika Anda ingin pekerjaan Flink keluar setelah semua log dikonsumsi, Anda juga harus mengatur exitAfterFinish ke true.
  consumerGroup	Nama kelompok konsumen.	String	Tidak	None	Kelompok konsumen digunakan untuk mencatat progres konsumsi. Anda dapat menentukan nama kustom tanpa format tetap. Catatan Pekerjaan Flink yang berbeda harus menggunakan kelompok konsumen yang berbeda. Jika beberapa pekerjaan Flink menggunakan kelompok konsumen yang sama, mereka tidak berkoordinasi dan masing-masing pekerjaan mengonsumsi semua data. Hal ini karena Flink tidak menggunakan kelompok konsumen SLS untuk penugasan partisi saat mengonsumsi data dari SLS. Akibatnya, setiap konsumen mengonsumsi pesan secara independen, meskipun mereka berbagi kelompok konsumen yang sama.
  consumeFromCheckpoint	Menentukan apakah akan mengonsumsi dari checkpoint kelompok konsumen.	String	Tidak	false	true: Anda juga harus menentukan kelompok konsumen. Program Flink mulai mengonsumsi log dari checkpoint yang disimpan dalam kelompok konsumen. Jika kelompok konsumen tidak memiliki checkpoint yang sesuai, konsumsi dimulai dari nilai konfigurasi startTime. false (nilai default): Tidak mulai mengonsumsi log dari checkpoint yang disimpan untuk kelompok konsumen yang ditentukan. Penting Parameter ini tidak lagi didukung di VVR 11.1 dan versi yang lebih baru. Untuk versi tersebut, Anda harus mengatur opsi startupMode ke consumer_group.
  maxRetries	Jumlah percobaan ulang setelah upaya membaca dari SLS gagal.	String	Tidak	3	N/A
  batchGetSize	Jumlah kelompok log yang dibaca dalam satu permintaan.	String	Tidak	100	Pengaturan batchGetSize tidak boleh melebihi 1.000. Jika dilanggar, sistem akan melaporkan error.
  exitAfterFinish	Menentukan apakah pekerjaan Flink keluar setelah semua data dikonsumsi.	String	Tidak	false	true: Program Flink keluar setelah semua data dikonsumsi. false (default): Program Flink tidak keluar setelah konsumsi data selesai.
  query Penting Opsi ini sudah tidak digunakan lagi di VVR 11.3, tetapi versi yang lebih baru tetap kompatibel.	Pernyataan kueri untuk preprocessing data sebelum konsumsi.	String	Tidak	None	Gunakan opsi ini untuk memfilter data SLS sebelum Flink mengonsumsinya. Hal ini mengurangi biaya dan meningkatkan kecepatan pemrosesan. Misalnya, 'query' = '*| where request_method = ''GET''' menunjukkan bahwa sebelum Flink membaca data dari SLS, sistem terlebih dahulu mencocokkan data di mana nilai bidang request_method adalah 'GET'. Catatan Opsi ini menggunakan bahasa SPL Log Service (SLS). Untuk informasi lebih lanjut, lihat SPL syntax. Penting Opsi ini hanya didukung di VVR 8.0.1 dan versi yang lebih baru. Fitur ini dikenai biaya Log Service (SLS). Untuk informasi lebih lanjut, lihat Billing of Log Service.
  processor	Nama prosesor SLS untuk preprocessing data. Jika query dan processor keduanya ditentukan, query memiliki prioritas lebih tinggi dan processor diabaikan.	String	Tidak	None	Opsi ini memfilter data SLS sebelum Flink mengonsumsinya, yang mengurangi biaya dan meningkatkan kecepatan pemrosesan. Kami menyarankan agar Anda menggunakan processor daripada query. Misalnya, 'processor' = 'test-filter-processor' menunjukkan bahwa prosesor SLS memfilter data sebelum Flink membaca data dari SLS. Catatan Opsi ini menggunakan bahasa SPL Log Service (SLS). Untuk informasi lebih lanjut, lihat SPL syntax. Untuk informasi tentang cara membuat atau memperbarui prosesor SLS, lihat Manage processors. Penting Opsi ini hanya didukung di VVR 11.3 dan versi yang lebih baru. Fitur ini dikenai biaya Log Service (SLS). Untuk informasi lebih lanjut, lihat Billing of Log Service.

• Khusus sink

  Parameter	Deskripsi	Tipe	Wajib	Default	Keterangan
  topicField	Menentukan bidang yang nilainya menimpa atribut __topic__, yang menunjukkan topik log.	String	Tidak	None	Nilai opsi ini harus merupakan bidang yang ada dalam tabel.
  timeField	Menentukan bidang yang nilainya menimpa atribut __timestamp__, yang menunjukkan waktu penulisan log.	String	Tidak	Waktu saat ini	Nilai opsi ini harus merupakan bidang INT yang ada dalam tabel. Jika opsi ini tidak ditentukan, waktu saat ini digunakan.
  sourceField	Menentukan bidang yang nilainya menimpa atribut __source__, yang menunjukkan sumber log, seperti alamat IP mesin yang menghasilkan log.	String	Tidak	None	Nilai opsi ini harus merupakan bidang yang ada dalam tabel.
  partitionField	Menentukan bidang untuk partisi. Hash dari nilai bidang ini menentukan shard mana yang menerima data, memastikan catatan dengan hash yang sama masuk ke shard yang sama.	String	Tidak	None	Jika opsi ini tidak ditentukan, setiap catatan ditulis secara acak ke shard yang tersedia.
  buckets	Jika partitionField ditentukan, opsi ini menentukan jumlah bucket untuk memetakan nilai hash.	String	Tidak	64	Nilai harus merupakan pangkat 2 dalam rentang [1, 256]. Jumlah bucket harus lebih besar dari atau sama dengan jumlah shard. Jika tidak, beberapa shard mungkin tidak menerima data apa pun.
  flushIntervalMs	Interval penulisan data.	String	Tidak	2000	Satuan: milidetik.
  writeNullProperties	Menentukan apakah nilai null ditulis sebagai string kosong ke SLS.	Boolean	Tidak	true	true (nilai default): Menulis nilai null ke log sebagai string kosong. false: Bidang yang bernilai null tidak ditulis ke log. Catatan Opsi ini hanya didukung di VVR 8.0.6 dan versi yang lebih baru.

Pemetaan tipe

Batasan

Fitur ini hanya didukung oleh Realtime Compute for Apache Flink versi 11.1 dan yang lebih baru.

Sintaksis

source:
   type: sls
   name: SLS Source
   endpoint: <endpoint>
   project: <project>
   logstore: <logstore>
   accessId: <accessId>
   accessKey: <accessKey>

Parameter

Gunakan katalog yang sudah ada

Mulai dari Realtime Compute for Apache Flink versi 11.5, Anda dapat mereferensikan SLS Catalog bawaan yang dibuat di halaman Data Management dalam pekerjaan data ingestion Flink CDC. Hal ini menghindari kebutuhan untuk menentukan properti koneksi secara manual.

source:
  type: sls
  using.built-in-catalog: sls_catalog

Saat ini, pekerjaan data ingestion dapat secara otomatis menggunakan kembali parameter berikut dari SLS Catalog bawaan:

• endpoint

• project

• accessId

• accessKey

Untuk mengganti parameter yang digunakan kembali secara otomatis ini, Anda dapat mendefinisikannya secara eksplisit dalam konfigurasi YAML. Parameter yang didefinisikan dalam file YAML memiliki prioritas lebih tinggi.

Pemetaan tipe data

Jika fixed-types tidak dikonfigurasi, pemetaan tipe data berikut berlaku:

Jika fixed-types dikonfigurasi, sistem mengurai data dengan menggunakan tipe yang ditentukan.

Inferensi dan evolusi skema

• Pre-konsumsi data shard dan inisialisasi skema

  Konektor SLS mempertahankan skema Logstore yang sedang dibacanya. Sebelum membaca data dari Logstore, konektor melakukan pre-consume hingga maxPreFetchLogGroups kelompok log dari setiap shard. Konektor mengurai skema setiap entri log dan menggabungkannya untuk menginisialisasi skema tabel. Event pembuatan tabel kemudian dihasilkan berdasarkan skema awal ini sebelum konsumsi data dimulai.

  Catatan

  Untuk setiap shard, konektor mencoba mengonsumsi data mulai dari satu jam sebelum waktu saat ini untuk mengurai skema log.

• Informasi primary key

  Log Log Service (SLS) tidak berisi informasi primary key. Anda dapat menambahkan primary key secara manual ke tabel dengan menggunakan aturan transformasi:

  transform:
    - source-table: <project>.<logstore>
      projection: *
      primary-keys: key1, key2

• Inferensi skema dan perubahan skema

  Setelah skema diinisialisasi, jika schema.inference.strategy diatur ke static, konektor SLS mengurai setiap entri log berdasarkan skema awal dan tidak menghasilkan event perubahan skema. Jika schema.inference.strategy diatur ke continuous, konektor mengurai setiap entri log, menginferensi kolom fisik, dan membandingkannya dengan skema saat ini. Jika skema yang diinferensi tidak konsisten dengan skema saat ini, skema digabungkan sesuai aturan berikut:

  • Jika kolom fisik yang diinferensi berisi bidang yang tidak ada dalam skema saat ini, konektor menambahkan bidang tersebut ke skema dan menghasilkan event untuk menambahkan kolom nullable.

  • Jika kolom fisik yang diinferensi tidak memiliki bidang yang ada dalam skema saat ini, konektor mempertahankan bidang tersebut, mengisi datanya dengan NULL, dan tidak menghasilkan event penghapusan kolom.

  Konektor SLS menginferensi tipe data semua bidang dalam setiap entri log sebagai String. Saat ini, hanya penambahan kolom baru yang didukung. Konektor menambahkan kolom baru di akhir skema dan mengaturnya sebagai nullable.