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

dengan parameter

• Parameter umum

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Catatan
  connector	Jenis tabel.	String	Ya	Tidak ada	Nilai tetap: sls.
  endPoint	Alamat titik akhir.	String	Ya	Tidak ada	Masukkan titik akhir jaringan pribadi untuk SLS. Untuk informasi selengkapnya, lihat Titik akhir layanan. Catatan Realtime Compute for Apache Flink tidak mendukung akses jaringan publik secara default. Namun, Alibaba Cloud NAT Gateway memungkinkan komunikasi antara jaringan VPC dan jaringan publik. Untuk informasi selengkapnya, lihat Bagaimana cara mengakses jaringan publik?. Hindari mengakses SLS melalui jaringan publik. Jika diperlukan, gunakan HTTPS dan aktifkan Global Accelerator (GA) untuk SLS. Untuk informasi selengkapnya, lihat Kelola akselerasi transfer.
  project	Nama proyek SLS.	String	Ya	Tidak ada	Tidak ada.
  logStore	Nama Logstore atau metricstore SLS.	String	Ya	Tidak ada	Logstore dan metricstore menggunakan metode konsumsi yang sama.
  accessId	ID AccessKey Akun Alibaba Cloud Anda.	String	Ya	Tidak ada	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan rahasia AccessKey saya?. Penting Untuk menghindari paparan Informasi AccessKey Anda, gunakan variabel untuk menentukan nilai AccessKey. Untuk informasi selengkapnya, lihat Variabel proyek.
  accessKey	Rahasia AccessKey Akun Alibaba Cloud Anda.	String	Ya	Tidak ada

• Parameter khusus sumber

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Catatan
  enableNewSource	Apakah akan mengaktifkan sumber baru yang mengimplementasikan antarmuka FLIP-27.	Boolean	Tidak	false	Sumber baru secara otomatis menyesuaikan perubahan shard dan mendistribusikan shard secara merata ke semua task sumber. Penting Parameter ini didukung pada VVR 8.0.9 dan versi yang lebih baru. Mulai dari VVR 11.1, nilai default-nya adalah true. Jika Anda mengubah parameter ini, pekerjaan tidak dapat dilanjutkan dari status sebelumnya. Untuk melanjutkan konsumsi dari offset historis, pertama-tama mulai pekerjaan dengan parameter consumerGroup untuk mencatat progres konsumsi dalam kelompok konsumen SLS. Kemudian atur consumeFromCheckpoint ke true dan restart pekerjaan tanpa state. Jika terdapat shard read-only di SLS, beberapa task Flink mungkin terus meminta shard lain yang belum diproses setelah menyelesaikan konsumsi shard read-only. Hal ini dapat menyebabkan distribusi shard yang tidak merata di antara task konkuren, sehingga mengurangi efisiensi konsumsi keseluruhan dan kinerja sistem. Untuk mengatasi masalah ini, sesuaikan konkurensi, optimalkan penjadwalan task, atau gabungkan shard kecil untuk mengurangi jumlah shard dan kompleksitas penugasan task.
  shardDiscoveryIntervalMs	Interval untuk mendeteksi perubahan shard secara dinamis, dalam milidetik.	Long	Tidak	60000	Atur ke nilai negatif untuk menonaktifkan deteksi dinamis. Catatan Nilai ini harus minimal 60000 ms (1 menit). Parameter ini hanya berlaku ketika enableNewSource bernilai true. Parameter ini didukung pada 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 tertentu. latest: Mengonsumsi log mulai dari offset terbaru. earliest: Mengonsumsi log mulai dari offset paling awal. consumer_group: Mengonsumsi log mulai dari offset yang dicatat dalam kelompok konsumen. Jika tidak ada offset yang dicatat untuk suatu shard, konsumsi dimulai dari offset paling awal. Penting Versi VVR sebelum 11.1 tidak mendukung consumer_group. Atur consumeFromCheckpoint ke true. Konsumsi kemudian dimulai dari offset yang dicatat dalam kelompok konsumen yang ditentukan, dan startupMode tidak berpengaruh.
  startTime	Waktu mulai untuk mengonsumsi log.	String	Tidak	Waktu saat ini	Format: yyyy-MM-dd hh:mm:ss. Hanya berlaku ketika startupMode diatur ke timestamp. Catatan startTime dan stopTime didasarkan pada atribut __receive_time__ di SLS, bukan __timestamp__.
  stopTime	Waktu akhir untuk mengonsumsi log.	String	Tidak	Tidak ada	Format: yyyy-MM-dd hh:mm:ss. Catatan Gunakan parameter ini hanya untuk mengonsumsi log historis. Atur ke waktu lampau. Jika diatur ke waktu mendatang, konsumsi mungkin berakhir lebih awal karena kurangnya log baru, menyebabkan gangguan aliran data tanpa pesan error. Untuk menghentikan program Flink ketika konsumsi log selesai, atur juga exitAfterFinish=true.
  consumerGroup	Nama kelompok konsumen.	String	Tidak	Tidak ada	Kelompok konsumen mencatat progres konsumsi. Anda dapat menentukan nama kustom tanpa batasan. Catatan Anda tidak dapat mengoordinasikan konsumsi lintas beberapa pekerjaan menggunakan kelompok konsumen yang sama. Setiap pekerjaan Flink harus menggunakan kelompok konsumen yang unik. Jika beberapa pekerjaan berbagi kelompok konsumen yang sama, mereka akan mengonsumsi semua data. Hal ini terjadi karena Flink tidak menetapkan partisi melalui kelompok konsumen SLS, sehingga setiap konsumen memproses pesan secara independen terlepas dari kelompok yang dibagikan.
  consumeFromCheckpoint	Apakah akan mengonsumsi log mulai dari checkpoint yang disimpan dalam kelompok konsumen yang ditentukan.	String	Tidak	false	true: Tentukan kelompok konsumen. Program Flink mengonsumsi log mulai dari checkpoint yang disimpan dalam kelompok tersebut. Jika tidak ada checkpoint, konsumsi dimulai dari nilai startTime. false (default): Jangan mengonsumsi log mulai dari checkpoint yang disimpan dalam kelompok konsumen yang ditentukan. Penting VVR 11.1 dan versi yang lebih baru tidak mendukung parameter ini. Untuk VVR 11.1 dan versi yang lebih baru, atur startupMode ke consumer_group.
  maxRetries	Jumlah percobaan ulang setelah pembacaan SLS gagal.	String	Tidak	3	Tidak ada.
  batchGetSize	Jumlah kelompok log yang dibaca per permintaan.	String	Tidak	100	Nilai batchGetSize tidak boleh melebihi 1000. Jika tidak, terjadi error.
  exitAfterFinish	Apakah program Flink keluar setelah konsumsi data selesai.	String	Tidak	false	true: Program Flink keluar setelah konsumsi data selesai. false (default): Program Flink terus berjalan setelah konsumsi data selesai.
  query Penting Usang mulai dari VVR 11.3. Masih kompatibel di versi yang lebih baru.	Pernyataan pra-pemrosesan konsumsi SLS.	String	Tidak	Tidak ada	Gunakan parameter query untuk memfilter data SLS sebelum dikonsumsi. Hal ini menghindari konsumsi semua data ke Flink, sehingga menghemat biaya dan meningkatkan kecepatan pemrosesan. Misalnya, 'query' = '*| where request_method = ''GET''' mencocokkan log yang bidang request_method-nya sama dengan GET sebelum Flink membacanya. Catatan Gunakan sintaksis SPL untuk kueri. Untuk informasi selengkapnya, lihat Sintaksis SPL. Penting Parameter ini didukung pada VVR 8.0.1 dan versi yang lebih baru. Fitur ini dikenai biaya SLS. Untuk informasi selengkapnya, lihat Harga.
  processor	Prosesor konsumen SLS. Memiliki prioritas lebih tinggi daripada query jika keduanya ada.	String	Tidak	Tidak ada	Gunakan parameter processor untuk memfilter data SLS sebelum dikonsumsi. Hal ini menghindari konsumsi semua data ke Flink, sehingga menghemat biaya dan meningkatkan kecepatan pemrosesan. Kami merekomendasikan processor daripada query. Misalnya, 'processor' = 'test-filter-processor' menerapkan prosesor konsumen SLS sebelum Flink membaca data. Catatan Gunakan sintaksis SPL untuk prosesor. Untuk informasi selengkapnya, lihat Sintaksis SPL. Untuk petunjuk membuat atau memperbarui prosesor konsumen, lihat Kelola prosesor konsumen. Penting Parameter ini didukung pada VVR 11.3 dan versi yang lebih baru. Fitur ini dikenai biaya SLS. Untuk informasi selengkapnya, lihat Harga.

• Hanya untuk tabel sink

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Catatan
  topicField	Nama bidang yang nilainya menggantikan bidang metadata __topic__. Mewakili topik log.	String	Tidak	Tidak ada	Bidang ini harus ada dalam tabel.
  timeField	Nama bidang yang nilainya menggantikan bidang metadata __timestamp__. Mewakili waktu penulisan log.	String	Tidak	Waktu saat ini	Bidang ini harus ada dalam tabel dan memiliki tipe data INT. Jika tidak ditentukan, waktu saat ini digunakan.
  sourceField	Nama bidang yang nilainya menggantikan bidang metadata __source__. Mewakili sumber log, seperti alamat IP mesin yang menghasilkan log.	String	Tidak	Tidak ada	Bidang ini harus ada dalam tabel.
  partitionField	Nama bidang. Data ditulis ke shard berdasarkan nilai hash bidang ini. Data dengan nilai hash yang sama masuk ke shard yang sama.	String	Tidak	Tidak ada	Jika tidak ditentukan, setiap catatan ditulis secara acak ke shard yang tersedia.
  buckets	Jumlah bucket untuk menetapkan ulang data ketika partitionField ditentukan.	String	Tidak	64	Nilai valid: bilangan bulat dari 1 hingga 256 yang merupakan pangkat dua. Jumlah bucket harus lebih besar dari atau sama dengan jumlah shard. Jika tidak, beberapa shard tidak menerima data.
  flushIntervalMs	Interval waktu yang memicu penulisan data.	String	Tidak	2000	Unit: milidetik.
  writeNullProperties	Apakah akan menulis nilai null sebagai string kosong ke SLS.	Boolean	Tidak	true	true (default): Menulis nilai null sebagai string kosong. false: Melewati bidang dengan nilai null saat menulis. Catatan Parameter ini didukung pada VVR 8.0.6 dan versi yang lebih baru.

Pemetaan tipe

Batasan

Fitur ini hanya didukung pada VVR 11.1 dan versi yang lebih baru.

Sintaksis

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

Item konfigurasi

Pemetaan tipe

Ketika fixed-types tidak dikonfigurasi, pemetaan tipe data ingestion adalah sebagai berikut:

Ketika fixed-types dikonfigurasi, konektor mencoba mengurai data menggunakan tipe yang ditentukan.

Inferensi skema dan sinkronisasi perubahan skema

• Pra-konsumsi shard dan inisialisasi skema

  Konektor SLS mempertahankan skema Logstore saat ini dari mana ia membaca data. Sebelum membaca data, konektor melakukan pra-konsumsi hingga `maxPreFetchLogGroups` kelompok log per shard. Kemudian, konektor mengurai skema setiap log dan menggabungkan skema tersebut untuk menginisialisasi struktur tabel. Selanjutnya, sebelum konsumsi aktual, konektor menghasilkan event pembuatan tabel yang sesuai berdasarkan skema yang diinisialisasi.

  Catatan

  Untuk setiap shard, konektor SLS mencoba mengonsumsi dan mengurai skema log mulai satu jam sebelum waktu saat ini.

• Informasi kunci primer

  Log SLS tidak berisi informasi kunci primer. Anda dapat menambahkan kunci primer secara manual menggunakan aturan transformasi:

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

• Inferensi skema dan perubahan skema

  Setelah inisialisasi skema, jika schema.inference.strategy diatur ke static, konektor SLS mengurai setiap log menggunakan skema awal dan tidak menghasilkan event perubahan skema. Jika schema.inference.strategy diatur ke continuous, konektor SLS mengurai setiap log, melakukan inferensi kolom fisik, dan membandingkannya dengan skema saat ini. Konektor kemudian menggabungkan skema ketika terdapat perbedaan. Aturan penggabungan:

  • Jika kolom fisik yang diinferensi mencakup bidang yang tidak ada dalam skema saat ini, bidang tersebut ditambahkan ke skema dan event kolom nullable baru dihasilkan.

  • Jika kolom fisik yang diinferensi tidak mencakup bidang yang sudah ada dalam skema saat ini, bidang tersebut tetap dipertahankan. Datanya diisi dengan NULL, dan tidak ada event penghapusan kolom yang dihasilkan.

  Konektor SLS melakukan inferensi semua tipe bidang dalam log sebagai `STRING`. Saat ini, hanya penambahan kolom yang didukung. Kolom baru ditambahkan di akhir skema saat ini dan diatur sebagai nullable.