Realtime Compute for Apache Flink：Konektor 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}'
);

Opsi konektor dalam klausa WITH

• Umum

  Opsi	Deskripsi	Tipe data	Diperlukan?	Nilai default	Catatan
  connector	Konektor yang akan digunakan.	String	Ya	Tidak ada nilai default	Atur ke sls.
  endPoint	Titik akhir SLS.	String	Ya	Tidak ada nilai default	Masukkan alamat akses VPC SLS. Untuk informasi lebih lanjut, lihat Titik akhir. Catatan Secara default, Realtime Compute for Apache Flink tidak dapat mengakses Internet. Namun, Alibaba Cloud menyediakan NAT gateway untuk memungkinkan komunikasi antara VPC dan Internet. Untuk informasi lebih lanjut, lihat Bagaimana Realtime Compute for Apache Flink mengakses Internet?. Kami sarankan agar Anda tidak mengakses SLS melalui Internet. Jika Anda perlu mengakses SLS melalui Internet, gunakan HTTPS dan aktifkan akselerasi transfer untuk SLS.
  project	Nama proyek SLS.	String	Ya	Tidak ada nilai default
  logStore	Nama Logstore atau Metricstore SLS.	STRING	Ya	Tidak ada nilai default	Data dalam Logstore dikonsumsi dengan cara yang sama seperti dalam Metricstore.
  accessId	ID AccessKey akun Alibaba Cloud Anda.	STRING	Ya	Tidak ada nilai default	Untuk informasi lebih lanjut, lihat Bagaimana cara melihat pasangan AccessKey suatu akun? Penting Untuk melindungi pasangan AccessKey Anda, konfigurasikan AccessKey Anda menggunakan variabel.
  accessKey	Rahasia AccessKey akun Alibaba Cloud Anda.	STRING	Ya	Tidak ada nilai default

• Spesifik Sumber

  Opsi	Deskripsi	Tipe data	Diperlukan?	Nilai default	Catatan
  enableNewSource	Menentukan apakah akan menggunakan antarmuka sumber refactor FLIP-27.	BOOLEAN	Tidak	false	Aktifkan opsi ini dan sumber secara otomatis menyesuaikan dengan perubahan shard dan mendistribusikan shard di seluruh subtugas sumber seimbang mungkin. Catatan Hanya VVR 8.0.9 atau lebih baru yang mendukung opsi ini. Penting Mulai dari VVR 11.1, opsi ini diatur ke true secara default. Jika nilai opsi berubah, pekerjaan Anda tidak dapat dilanjutkan dari status tertentu. Untuk menyelesaikan ini, konfigurasikan opsi consumerGroup untuk mencatat offset konsumen saat ini dan mulai pekerjaan Anda. Kemudian, atur consumeFromCheckpoint ke true dan mulai pekerjaan tanpa status. Saat subtugas sumber selesai membaca dari shard read-only, mereka terus meminta mengonsumsi shard lain. Hal ini dapat menyebabkan konsumsi shard yang tidak merata di antara subtugas sumber, memengaruhi kinerja keseluruhan pekerjaan. Untuk mengurangi masalah ini, pertimbangkan menyesuaikan paralelisme sumber, mengoptimalkan strategi penjadwalan Anda, atau menggabungkan shard kecil untuk menyederhanakan penugasan shard.
  shardDiscoveryIntervalMs	Interval deteksi dinamis perubahan shard.	LONG	Tidak	60000	Unit: milidetik. Untuk menonaktifkan deteksi dinamis, atur opsi ke nilai negatif. Catatan Nilai opsi ini tidak boleh kurang dari 1 menit (atau 60.000 milidetik). Opsi ini hanya berlaku jika opsi enableNewSource diatur ke true. Hanya VVR 8.0.9 atau lebih baru yang mendukung opsi ini.
  startupMode	Mode startup tabel sumber.	STRING	Tidak	timestamp	timestamp: Log dikonsumsi dari waktu mulai yang ditentukan. latest: Log dikonsumsi dari offset terbaru. earliest: Log dikonsumsi dari offset paling awal. consumer_group: Log dikonsumsi dari offset yang direkam dalam kelompok konsumen. Jika kelompok konsumen tidak mencatat offset konsumsi shard, log dikonsumsi dari offset paling awal. Penting Di versi VVR sebelum 11.1, consumer_group tidak lagi didukung. Untuk mengonsumsi data dari offset yang direkam dalam kelompok konsumen tertentu, atur consumeFromCheckpoint ke true.
  startTime	Waktu mulai konsumsi log.	STRING	Tidak	Waktu saat ini	Nilai opsi ini dalam format yyyy-MM-dd hh:mm:ss. Opsi ini berlaku hanya jika startupMode diatur ke timestamp. Catatan Parameter startTime dan stopTime dikonfigurasi berdasarkan bidang __receive_time__ dalam tabel sumber SLS, bukan pada bidang __timestamp__.
  stopTime	Waktu berhenti konsumsi log.	String	Tidak	Tidak ada nilai default	Nilai opsi ini dalam format yyyy-MM-dd hh:mm:ss. Catatan Untuk hanya mengonsumsi log historis, atur opsi ini ke titik waktu historis tertentu. Menggunakan titik waktu di masa depan dapat menyebabkan konsumsi berhenti secara tak terduga jika injeksi log baru sementara terganggu. Gejala yang teramati adalah gangguan aliran data tanpa pesan kesalahan atau pengecualian yang menyertainya. Jika Anda ingin program Realtime Compute for Apache Flink keluar setelah konsumsi log selesai, Anda juga harus mengonfigurasi opsi exitAfterFinish dan atur opsi exitAfterFinish ke true.
  consumerGroup	Nama grup konsumen.	STRING	Tidak	Tidak ada nilai default	Grup konsumen mencatat kemajuan konsumsi. Anda dapat menentukan nama grup konsumen kustom. Format nama tidak tetap. Catatan Grup konsumen tidak dapat dibagikan oleh beberapa pekerjaan untuk konsumsi kolaboratif. Kami sarankan Anda menentukan grup konsumen berbeda untuk pekerjaan berbeda. Jika Anda menentukan grup konsumen yang sama untuk pekerjaan berbeda, semua data dikonsumsi. Ketika Realtime Compute for Apache Flink mengonsumsi data dari SLS, data tidak di-shard dalam grup konsumen. Oleh karena itu, jika beberapa pekerjaan berbagi grup konsumen yang sama, semua pesan dalam grup konsumen dikonsumsi oleh setiap pekerjaan.
  consumeFromCheckpoint	Menentukan apakah akan mengonsumsi log dari checkpoint yang disimpan dalam grup konsumen yang ditentukan.	STRING	Tidak	false	true: Jika Anda mengatur opsi ini ke true, Anda juga harus menentukan grup konsumen. Flink mengonsumsi log dari titik kontrol yang disimpan dalam grup konsumen. Jika tidak ada titik kontrol dalam grup konsumen, Flink mengonsumsi log dari waktu yang ditentukan oleh opsi startTime. false : Flink tidak mengonsumsi log dari titik kontrol yang disimpan dalam grup konsumen tertentu. Penting Mulai dari VVR 11.1, opsi ini tidak lagi didukung. Anda perlu mengatur startupMode ke consumer_group.
  maxRetries	Jumlah percobaan ulang yang diizinkan ketika gagal membaca data dari SLS.	String	Tidak	3
  batchGetSize	Jumlah grup log yang dibaca dalam satu permintaan.	String	Tidak	100	Untuk mencegah kesalahan, atur batchGetSize ke nilai kurang dari 1000.
  exitAfterFinish	Menentukan apakah program Realtime Compute for Apache Flink keluar setelah konsumsi data selesai.	String	Tidak	false	true false
  query Penting Opsi ini sudah ditinggalkan di VVR 11.3, tetapi versi berikutnya tetap kompatibel.	Pernyataan query yang digunakan untuk memproses data sebelum konsumsi data dimulai.	STRING	Tidak	Tidak ada nilai default	Mengonfigurasi opsi ini untuk menyaring data dari SLS sebelum konsumsi data dimulai, mengurangi biaya dan meningkatkan efisiensi pemrosesan data. Sebagai contoh, jika Anda menentukan 'query' = '*| where request_method = ''GET''', Realtime Compute for Apache Flink menyaring data di mana nilai bidang request_method sama dengan GET sebelum konsumsi data. Catatan Gunakan sintaks SPL saat mengonfigurasi opsi ini. Penting Hanya VVR 8.0.1 atau lebih baru yang mendukung opsi ini. Fitur ini membebankan biaya dari SLS. Untuk detailnya, lihat Tagihan.
  processor	Prosesor SLS. Jika kedua opsi ini dan query diatur, maka query memiliki prioritas lebih tinggi.	STRING	Tidak	Tidak ada nilai default	Opsi ini secara fungsional setara dengan query, tetapi kami sarankan menggunakan opsi ini. Sebagai contoh, pengaturan 'processor' = 'test-filter-processor' menunjukkan bahwa data akan difilter oleh prosesor SLS sebelum dikonsumsi oleh Flink. Catatan Gunakan sintaksis SPL saat mengonfigurasi opsi ini. Penting Hanya VVR 8.0.1 atau lebih baru yang mendukung opsi ini. Fitur ini membebankan biaya dari SLS. Untuk detailnya, lihat Tagihan.

• Spesifik Sink

  Opsi	Deskripsi	Tipe data	Diperlukan?	Nilai default	Catatan
  topicField	Menentukan nama bidang. Nilai opsi ini menimpa nilai bidang __topic__ untuk menunjukkan topik log.	String	Tidak	Tidak ada nilai default	Nilai opsi ini harus berupa bidang yang ada dalam tabel.
  timeField	Menentukan nama bidang. Nilai opsi ini menimpa nilai bidang __timestamp__ untuk menunjukkan waktu penulisan log.	String	Tidak	Waktu saat ini	Opsi ini harus diatur ke bidang INT yang ada. Jika tidak ada bidang yang ditentukan, waktu saat ini digunakan.
  sourceField	Menentukan nama bidang. Nilai opsi ini menimpa nilai bidang atribut __source__ untuk menunjukkan asal log. Sebagai contoh, nilainya adalah alamat IP mesin yang menghasilkan log.	STRING	Tidak	Tidak ada nilai default	Nilai opsi ini harus berupa bidang yang ada dalam tabel.
  partitionField	Menentukan nama bidang. Nilai hash dihitung berdasarkan nilai parameter ini ketika data ditulis ke SLS. Data yang memiliki nilai hash yang sama ditulis ke shard yang sama.	STRING	Tidak	Tidak ada nilai default	Jika Anda tidak menentukan opsi ini, setiap entri data secara acak ditulis ke shard yang tersedia.
  buckets	Jumlah bucket yang dikelompokkan ulang berdasarkan nilai hash ketika opsi partitionField ditentukan.	String	Tidak	64	Rentang nilai valid: [1,256]. Nilai opsi ini harus merupakan pangkat dua bilangan bulat. Jumlah bucket harus lebih besar dari atau sama dengan jumlah shard. Jika tidak, tidak ada data yang akan ditulis ke shard tertentu.
  flushIntervalMs	Interval di mana penulisan data dipicu.	STRING	Tidak	2000	Unit: milidetik.
  writeNullProperties	Menentukan apakah akan menulis nilai null sebagai string kosong ke SLS.	BOOLEAN	Tidak	true	true false Catatan Hanya VVR 8.0.6 atau lebih baru yang mendukung opsi ini.

Pemetaan tipe data

Batasan

Hanya VVR 11.1 atau lebih baru yang mendukung pengambilan data dari SLS.

Sintaks

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

Opsi konfigurasi

Pemetaan tipe data

Pemetaan tipe data untuk pengambilan data adalah sebagai berikut:

Inferensi dan evolusi skema

• Konsumsi Data Awal dan Inisialisasi Skema

  Konektor SLS mempertahankan skema dari Logstore saat ini. Sebelum membaca data dari Logstore, konektor mencoba melakukan konsumsi awal hingga maxPreFetchLogGroups grup log dari setiap shard dan menginisialisasi skema dengan cara menguraikan dan menggabungkan skema dari setiap log. Selanjutnya, sebelum konsumsi data dimulai, sebuah event pembuatan tabel dihasilkan berdasarkan skema yang telah diinisialisasi.

  Catatan

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

• Kunci Utama

  Log SLS tidak mengandung kunci utama. Tambahkan secara manual kunci utama ke tabel di modul transform:

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

• Inferensi dan Evolusi Skema

  Setelah inisialisasi skema, jika schema.inference.strategy disetel ke static, konektor menguraikan setiap entri log berdasarkan skema dan tidak menghasilkan event perubahan skema. Jika schema.inference.strategy disetel ke continuous, konektor menguraikan setiap entri log, melakukan inferensi kolom fisik, dan membandingkannya dengan skema saat ini. Jika skema yang diinferensikan tidak konsisten dengan skema saat ini, skema digabungkan sesuai aturan berikut:

  • Jika skema yang diinferensikan mengandung kolom fisik yang tidak ada dalam skema saat ini, kolom yang hilang ditambahkan ke skema saat ini, dan event penambahan kolom nullable dihasilkan.

  • Jika skema yang diinferensikan tidak mengandung kolom tertentu dalam skema saat ini, kolom tersebut dipertahankan dan nilainya disetel ke NULL.

  Konektor SLS menginferensikan semua bidang sebagai bidang string. Saat ini, hanya penambahan kolom yang didukung. Kolom baru ditambahkan ke skema saat ini sebagai kolom nullable.