Realtime Compute for Apache Flink：Simple Log Service (SLS)

Sintaks

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}'
);

DENGAN parameter

• Umum

  Parameter	Deskripsi	Tipe data	Wajib?	Nilai default	Keterangan
  connector	Jenis tabel.	String	Ya	Tidak ada	Atur ke sls.
  endPoint	Alamat endpoint.	String	Ya	Tidak ada	Masukkan endpoint VPC SLS. Untuk informasi selengkapnya, lihat Endpoints. Catatan Secara default, Realtime Compute for Apache Flink tidak dapat mengakses Internet. Namun, Alibaba Cloud menyediakan gerbang NAT untuk mengaktifkan komunikasi antara VPC dan Internet. Untuk informasi selengkapnya, lihat Bagaimana cara mengakses Internet?. Hindari mengakses SLS melalui Internet. Jika harus melakukannya, gunakan HTTPS dan aktifkan transfer acceleration untuk SLS.
  project	Nama proyek SLS.	String	Ya	Tidak ada	Tidak ada.
  logStore	Nama Logstore atau Metricstore SLS.	String	Ya	Tidak ada	Data di Logstore dikonsumsi dengan cara yang sama seperti di Metricstore.
  accessId	ID AccessKey Akun Alibaba Cloud Anda.	String	Ya	Tidak ada nilai default	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan rahasia AccessKey?. Penting Untuk melindungi pasangan AccessKey Anda, gunakan variabel untuk mengonfigurasi AccessKey Anda.
  accessKey	Rahasia AccessKey Akun Alibaba Cloud Anda.	String	Ya	Tidak ada nilai default

• Khusus sumber

  Parameter	Deskripsi	Tipe data	Wajib?	Nilai default	Keterangan
  enableNewSource	Menentukan apakah akan menggunakan antarmuka sumber baru yang mengimplementasikan FLIP-27.	Boolean	Tidak	false	Sumber baru menyesuaikan secara otomatis terhadap perubahan shard dan mendistribusikan shard secara merata di semua subtugas sumber. Penting Opsi ini hanya didukung di VVR 8.0.9 atau yang lebih baru. Mulai dari VVR 11.1, opsi ini secara default bernilai true. Jika Anda mengubah opsi ini, pekerjaan Anda tidak dapat dilanjutkan dari state yang disimpan. Sebagai solusi, pertama-tama jalankan pekerjaan Anda dengan opsi consumerGroup untuk mencatat offset konsumen saat ini. Kemudian, atur consumeFromCheckpoint ke true dan mulai ulang pekerjaan Anda tanpa state. Jika SLS berisi shard read-only, beberapa subtugas Flink mungkin selesai membaca dari shard tersebut lalu meminta shard lain yang belum dibaca. Hal ini dapat menyebabkan distribusi shard yang tidak merata di antara subtugas, sehingga mengurangi efisiensi konsumsi dan kinerja sistem secara keseluruhan. Untuk mengurangi ketidakseimbangan ini, sesuaikan parallelisme sumber, optimalkan penjadwalan tugas, atau gabungkan shard kecil.
  shardDiscoveryIntervalMs	Interval deteksi dinamis perubahan shard. Satuan: milidetik.	Long	Tidak	60000	Atur opsi ini ke nilai negatif untuk menonaktifkan deteksi dinamis. Catatan Opsi ini harus minimal 1 menit (60.000 milidetik). Opsi ini hanya berlaku jika enableNewSource diatur ke true. Opsi ini hanya didukung di VVR 8.0.9 atau yang lebih baru.
  startupMode	Mode startup tabel sumber.	String	Tidak	timestamp	timestamp (default): Mengonsumsi log mulai dari waktu yang ditentukan. latest: Mengonsumsi log mulai dari offset terbaru. earliest: Mengonsumsi log mulai dari offset paling awal. consumer_group: Mengonsumsi log mulai dari offset yang direkam dalam kelompok konsumen. Jika tidak ada offset yang direkam untuk suatu shard, konsumsi log dimulai dari offset paling awal. Penting Pada versi VVR sebelum 11.1, nilai consumer_group tidak didukung. Untuk mengonsumsi log dari offset yang direkam oleh kelompok konsumen tertentu, atur consumeFromCheckpoint ke true. Dalam kasus ini, mode startup ini tidak berlaku.
  startTime	Waktu mulai mengonsumsi log.	String	Tidak	Waktu saat ini	Format: yyyy-MM-dd hh:mm:ss. Opsi ini hanya berlaku jika startupMode diatur ke timestamp. Catatan Opsi startTime dan stopTime didasarkan pada bidang __receive_time__ di SLS, bukan bidang __timestamp__.
  stopTime	Waktu akhir konsumsi log.	String	Tidak	Tidak ada	Format: yyyy-MM-dd hh:mm:ss. Catatan Gunakan opsi ini hanya untuk mengonsumsi log historis. Atur ke titik waktu di masa lalu. Jika Anda mengaturnya ke waktu di masa depan, konsumsi mungkin berhenti secara tak terduga jika tidak ada log baru yang ditulis. Hal ini tampak sebagai aliran data yang terputus tanpa pesan error. Untuk keluar dari program Flink setelah konsumsi log selesai, atur juga exitAfterFinish ke true.
  consumerGroup	Nama kelompok konsumen.	String	Tidak	Tidak ada	Kelompok konsumen mencatat progres konsumsi. Anda dapat menentukan nama kustom apa pun. Catatan Anda tidak dapat berbagi kelompok konsumen di beberapa pekerjaan untuk konsumsi kolaboratif. Gunakan kelompok konsumen berbeda untuk pekerjaan berbeda. Jika Anda menggunakan kelompok konsumen yang sama untuk pekerjaan berbeda, setiap pekerjaan akan mengonsumsi semua data. Saat Flink mengonsumsi data dari SLS, shard tidak ditugaskan melalui kelompok konsumen SLS. Oleh karena itu, setiap pekerjaan secara independen mengonsumsi semua pesan, meskipun menggunakan kelompok konsumen yang sama.
  consumeFromCheckpoint	Menentukan apakah akan mengonsumsi log dari checkpoint yang disimpan dalam kelompok konsumen tertentu.	String	Tidak	false	true: Jika Anda mengatur parameter ini ke true, Anda juga harus menentukan kelompok konsumen. Flink mengonsumsi log dari checkpoint yang disimpan dalam kelompok konsumen. Jika tidak ada checkpoint di kelompok konsumen, Flink mengonsumsi log dari waktu yang ditentukan oleh parameter startTime. false (default): Flink tidak mengonsumsi log dari checkpoint yang disimpan dalam kelompok konsumen tertentu. Penting Opsi ini tidak didukung di VVR 11.1 atau yang lebih baru. Untuk VVR 11.1 atau yang lebih baru, atur startupMode ke consumer_group.
  maxRetries	Jumlah percobaan ulang setelah pembacaan dari SLS gagal.	String	Tidak	3	Tidak ada.
  batchGetSize	Jumlah kelompok log yang dibaca per permintaan.	String	Tidak	100	Atur batchGetSize ke nilai kurang dari 1000. Jika tidak, terjadi error.
  exitAfterFinish	Menentukan apakah program Flink keluar setelah konsumsi data selesai.	String	Tidak	false	true: Program Flink keluar setelah konsumsi data selesai. false (default): Program Flink tidak keluar setelah konsumsi data selesai.
  query Penting Opsi ini sudah ditinggalkan di VVR 11.3 tetapi tetap kompatibel di versi yang lebih baru.	Pernyataan kueri yang digunakan untuk memproses data sebelum mengonsumsi data SLS.	String	Tidak	Tidak ada nilai default	Gunakan opsi query untuk memfilter data SLS sebelum dikonsumsi. Hal ini menghindari pemuatan semua data ke Flink, sehingga mengurangi biaya dan meningkatkan kecepatan pemrosesan. Contohnya, 'query' = '*| where request_method = ''GET''' memfilter log di mana bidang request_method bernilai GET sebelum Flink membacanya. Catatan Tulis kueri menggunakan sintaks SPL. Penting Opsi ini hanya didukung di VVR 8.0.1 atau yang lebih baru. Fitur ini dikenai biaya SLS. Untuk detailnya, lihat Tagihan.
  processor	Prosesor konsumen SLS. Jika query dan processor keduanya diatur, query memiliki prioritas lebih tinggi.	String	Tidak	Tidak ada	Gunakan opsi processor untuk memfilter data SLS sebelum dikonsumsi. Hal ini menghindari pemuatan semua data ke Flink, sehingga mengurangi biaya dan meningkatkan kecepatan pemrosesan. Kami merekomendasikan menggunakan processor daripada query. Contohnya, 'processor' = 'test-filter-processor' menerapkan prosesor konsumen SLS untuk memfilter data sebelum Flink membacanya. Catatan Tulis processor menggunakan sintaks SPL. Untuk detail tentang membuat dan memperbarui prosesor konsumen SLS, lihat Kelola prosesor konsumen. Penting Opsi ini hanya didukung di VVR 11.3 atau yang lebih baru. Fitur ini dikenai biaya SLS. Untuk detailnya, lihat Tagihan.

• Khusus sink

  parameter	Deskripsi	Tipe data	Wajib?	Nilai default	Keterangan
  topicField	Nama bidang yang nilainya menggantikan bidang __topic__. Ini menunjukkan topik log.	String	Tidak	Tidak ada	Parameter ini menentukan bidang yang sudah ada di tabel.
  timeField	Nama bidang yang nilainya menggantikan bidang __timestamp__. Ini menunjukkan waktu penulisan log.	String	Tidak	Waktu saat ini	Bidang ini harus ada di tabel dan tipenya harus INT. Jika tidak ditentukan, digunakan waktu saat ini.
  sourceField	Nama bidang yang nilainya menggantikan bidang __source__. Ini menunjukkan sumber log, seperti alamat IP mesin yang menghasilkan log.	String	Tidak	Tidak ada	Bidang ini harus ada di tabel.
  partitionField	Nama bidang. Nilai hash dihitung dari nilai bidang ini saat menulis data. Data dengan nilai hash yang sama ditulis ke shard yang sama.	String	Tidak	Tidak ada nilai default	Jika tidak ditentukan, setiap entri data ditulis secara acak ke shard yang tersedia.
  buckets	Jumlah bucket untuk dikelompokkan ulang berdasarkan nilai hash saat partitionField ditentukan.	String	Tidak	64	Nilai valid: [1, 256]. Nilainya harus merupakan pangkat dari 2. Jumlah bucket harus lebih besar dari atau sama dengan jumlah shard. Jika tidak, beberapa shard tidak menerima data.
  flushIntervalMs	Interval pemicuan penulisan data.	String	Tidak	2000	Satuan: milidetik.
  writeNullProperties	Menentukan apakah nilai null ditulis sebagai string kosong ke SLS.	Boolean	Tidak	true	true (default): Menulis nilai null sebagai string kosong. false: Tidak menulis bidang yang nilainya dihitung sebagai null. Catatan Opsi ini hanya didukung di VVR 8.0.6 atau yang lebih baru.

Pemetaan tipe data

Batasan

Hanya didukung oleh mesin komputasi waktu nyata Ververica Runtime (VVR) 11.1 dan yang lebih baru.

Sintaks

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

Opsi konfigurasi

Pemetaan tipe data

Pemetaan tipe data untuk data ingestion adalah sebagai berikut:

Inferensi dan evolusi skema

• Pre-konsumsi dan inisialisasi skema

  Konektor SLS mempertahankan skema Logstore saat ini. Sebelum membaca data dari Logstore, konektor melakukan pre-consume hingga maxPreFetchLogGroups kelompok log dari setiap shard, mengurai skema setiap log, lalu menggabungkannya untuk menginisialisasi skema tabel. Sebelum konsumsi aktual dimulai, konektor menghasilkan event pembuatan tabel berdasarkan skema yang diinisialisasi.

  Catatan

  Untuk setiap shard, konektor mencoba mengonsumsi data satu jam sebelum waktu saat ini guna mengurai skema.

• Kunci primer

  Log SLS tidak berisi kunci primer. Tambahkan kunci primer secara manual menggunakan aturan transformasi:

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

• Inferensi dan evolusi skema

  Setelah inisialisasi skema, jika schema.inference.strategy diatur ke static, konektor mengurai setiap entri log menggunakan skema awal dan tidak menghasilkan event perubahan skema. Jika schema.inference.strategy diatur ke continuous, konektor mengurai setiap entri log, menginferensi kolom fisik, lalu membandingkannya dengan skema saat ini. Jika skema yang diinferensi berbeda dari skema saat ini, konektor menggabungkannya menggunakan aturan berikut:

  • Jika skema yang diinferensi mencakup bidang yang tidak ada di skema saat ini, tambahkan bidang tersebut ke skema saat ini dan hasilkan event penambahan kolom nullable.

  • Jika skema yang diinferensi tidak mencakup bidang yang ada di skema saat ini, pertahankan bidang tersebut dan atur nilainya ke NULL. Jangan hasilkan event penghapusan kolom.

  Konektor SLS menginferensi semua bidang sebagai string. Saat ini, hanya penambahan kolom yang didukung. Kolom baru ditambahkan di akhir skema saat ini dan ditandai sebagai nullable.