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	Tipe	Wajib	Default	Keterangan
  connector	Konektor yang digunakan.	String	Ya	Tidak ada	Atur ke sls.
  endPoint	Titik akhir Log Service (SLS).	String	Ya	Tidak ada	Tentukan alamat akses VPC Log Service (SLS). Untuk informasi selengkapnya, lihat Titik akhir layanan. 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 selengkapnya, lihat Bagaimana cara mengakses Internet?. Kami menyarankan agar Anda tidak mengakses SLS melalui internet. Jika harus melakukannya, gunakan HTTPS dan aktifkan akselerasi transfer. Untuk informasi selengkapnya, lihat Kelola akselerasi transfer.
  project	Nama project SLS.	String	Ya	Tidak ada	N/A
  logStore	Nama Logstore atau MetricStore.	String	Ya	Tidak ada	Data dalam Logstore dikonsumsi dengan cara yang sama seperti data dalam MetricStore.
  accessId	ID AccessKey Akun Alibaba Cloud Anda.	String	Ya	Tidak ada	Untuk informasi selengkapnya, lihat Dapatkan Pasangan Kunci Akses. Penting Untuk mencegah Pasangan Kunci Akses Anda terpapar, kami menyarankan agar Anda menggunakan variabel untuk menentukan ID AccessKey dan rahasia AccessKey. Untuk informasi selengkapnya, lihat Variabel project.
  accessKey	Rahasia AccessKey Akun Alibaba Cloud Anda.	String	Ya	Tidak ada

• 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 seimbang di semua subtask sumber. Penting Opsi ini hanya didukung di VVR 8.0.9 dan yang lebih baru. Nilai default adalah true untuk VVR 11.1 dan 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 di 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 dan 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 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 direkam oleh kelompok konsumen. Jika kelompok konsumen belum merekam 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 direkam 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	Tidak ada	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 berhenti setelah semua log dikonsumsi, Anda juga harus mengatur exitAfterFinish ke true.
  consumerGroup	Nama kelompok konsumen.	String	Tidak	Tidak ada	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 di 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 yang lebih baru. Untuk versi tersebut, Anda harus mengatur opsi startupMode ke consumer_group.
  maxRetries	Jumlah percobaan ulang setelah gagal membaca dari SLS.	String	Tidak	3	N/A
  batchGetSize	Jumlah kelompok log yang dibaca dalam satu permintaan.	String	Tidak	100	Pengaturan batchGetSize tidak boleh melebihi 1000, atau akan muncul error.
  exitAfterFinish	Menentukan apakah pekerjaan Flink berhenti setelah semua data dikonsumsi.	String	Tidak	false	true: Program Flink berhenti setelah semua data dikonsumsi. false (default): Program Flink tidak berhenti 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 pra-pemrosesan data sebelum konsumsi.	String	Tidak	Tidak ada	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, data terlebih dahulu dicocokkan dengan nilai bidang request_method yang bernilai 'GET'. Catatan Opsi ini menggunakan bahasa SPL Log Service (SLS). Untuk informasi selengkapnya, lihat Sintaksis SPL. Penting Opsi ini hanya didukung di VVR 8.0.1 dan yang lebih baru. Fitur ini dikenai biaya Log Service (SLS). Untuk informasi selengkapnya, lihat Penagihan Log Service.
  processor	Nama prosesor SLS untuk pra-pemrosesan data. Jika query dan processor keduanya ditentukan, query yang diutamakan dan processor diabaikan.	String	Tidak	Tidak ada	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 selengkapnya, lihat Sintaksis SPL. Untuk informasi tentang cara membuat atau memperbarui prosesor SLS, lihat Kelola prosesor. Penting Opsi ini hanya didukung di VVR 11.3 dan yang lebih baru. Fitur ini dikenai biaya Log Service (SLS). Untuk informasi selengkapnya, lihat Penagihan Log Service.
  preserveRawBytes	Menentukan apakah akan langsung mengambil dan menyimpan byte mentah yang dibawa oleh SLS ketika enableNewSource diatur ke true.	Boolean	Tidak	false	Ketika opsi ini diaktifkan, bidang BINARY/VARBINARY membaca byte[] mentah secara langsung, dan bidang CHAR/VARCHAR membuat data string dari byte mentah tersebut. Bidang lain tetap menggunakan logika konversi default. Perilaku ini konsisten dengan sumber lama. Catatan Opsi ini hanya didukung di VVR 11.8 dan yang lebih baru. Opsi ini hanya berlaku ketika enableNewSource diatur ke true.

• Khusus sink

  Parameter	Deskripsi	Tipe	Wajib	Default	Keterangan
  topicField	Menentukan bidang yang nilainya menimpa atribut __topic__, yang menunjukkan topik log.	String	Tidak	Tidak ada	Nilai opsi ini harus merupakan bidang yang ada di 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 di tabel. Jika opsi ini tidak ditentukan, waktu saat ini yang digunakan.
  sourceField	Menentukan bidang yang nilainya menimpa atribut __source__, yang menunjukkan sumber log, seperti alamat IP mesin yang menghasilkan log.	String	Tidak	Tidak ada	Nilai opsi ini harus merupakan bidang yang ada di 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	Tidak ada	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 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 Katalog SLS bawaan yang dibuat di halaman Data Management dalam pekerjaan data ingestion Flink CDC, sehingga tidak perlu 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 Katalog SLS 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, mengurai skema setiap entri log, lalu 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 guna mengurai skema log.

• Informasi kunci primer

  Log Log Service (SLS) tidak berisi informasi kunci primer. Anda dapat menambahkan kunci primer ke tabel secara manual 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, lalu membandingkannya dengan skema saat ini. Jika skema yang diinferensi tidak konsisten dengan skema saat ini, skema tersebut digabungkan sesuai aturan berikut:

  • Jika kolom fisik yang diinferensi berisi bidang yang tidak ada di 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 di 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 ke akhir skema dan mengaturnya sebagai nullable.