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	Catatan
  connector	Jenis tabel.	String	Ya	Tidak ada	Atur parameter ini ke sls.
  endPoint	Titik akhir.	String	Ya	Tidak ada	Masukkan endpoint jaringan pribadi SLS. Untuk informasi selengkapnya, lihat Endpoints. Catatan Secara default, Realtime Compute for Apache Flink tidak dapat mengakses jaringan publik. Namun, Alibaba Cloud menyediakan NAT Gateway untuk mengaktifkan komunikasi antara VPC dan jaringan publik. Untuk informasi selengkapnya, lihat Bagaimana cara mengakses jaringan publik?. Kami tidak menyarankan Anda mengakses SLS melalui jaringan publik. Jika Anda harus mengakses SLS melalui jaringan publik, gunakan protokol 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	Data dalam Logstore dikonsumsi dengan cara yang sama seperti dalam metricstore.
  accessId	ID AccessKey akun Alibaba Cloud Anda.	String	Ya	Tidak ada	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan Rahasia AccessKey?. Penting Untuk mencegah Informasi AccessKey Anda bocor, kami menyarankan agar Anda menggunakan variabel untuk menentukan Pasangan Kunci Akses. Untuk informasi selengkapnya, lihat Kelola variabel.
  accessKey	Rahasia AccessKey akun Alibaba Cloud Anda.	String	Ya	Tidak ada

• Khusus tabel source

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Catatan
  enableNewSource	Menentukan apakah akan mengaktifkan sumber data baru yang mengimplementasikan antarmuka FLIP-27.	Boolean	Tidak	false	Sumber data baru dapat secara otomatis beradaptasi terhadap perubahan shard dan memastikan bahwa shard didistribusikan secara merata di seluruh semua konkurensi source. Penting Parameter ini hanya didukung di Realtime Compute for Apache Flink VVR 8.0.9 atau yang lebih baru. Parameter ini diatur ke true secara default mulai dari VVR 11.1. Pekerjaan tidak dapat dipulihkan dari state setelah item konfigurasi ini diubah. Untuk mengatasi masalah ini, Anda dapat terlebih dahulu mengatur item konfigurasi consumerGroup untuk memulai pekerjaan dan mencatat progres konsumsi dalam kelompok konsumen SLS. Kemudian, atur item konfigurasi consumeFromCheckpoint ke true dan mulai pekerjaan secara tanpa status. Dengan cara ini, pekerjaan dapat melanjutkan konsumsi dari progres historis. Jika terdapat shard read-only di SLS, beberapa task Flink konkuren tetap meminta data dari shard lain yang belum selesai setelah mereka selesai mengonsumsi data dari shard read-only tersebut. Hal ini dapat menyebabkan beberapa task konkuren ditugaskan ke beberapa shard, sehingga menyebabkan ketidakseimbangan distribusi shard di antara mereka. Ketidakseimbangan ini memengaruhi efisiensi konsumsi keseluruhan dan kinerja sistem. Untuk mengurangi masalah ini, sesuaikan konkurensi, optimalkan kebijakan penjadwalan task, atau gabungkan shard kecil untuk mengurangi jumlah shard dan kompleksitas penugasan task.
  shardDiscoveryIntervalMs	Interval deteksi dinamis perubahan shard. Satuan: milidetik.	Long	Tidak	60000	Atur parameter ini ke nilai negatif untuk menonaktifkan deteksi dinamis. Catatan Nilai parameter ini tidak boleh kurang dari 1 menit (60.000 milidetik). Parameter ini hanya berlaku ketika parameter enableNewSource diatur ke true. Parameter ini hanya didukung di Realtime Compute for Apache Flink VVR 8.0.9 atau yang lebih baru.
  startupMode	Mode startup tabel sumber.	String	Tidak	timestamp	timestamp (default): 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 dicatat dalam kelompok konsumen. Jika kelompok konsumen tidak mencatat offset konsumen suatu shard, log dikonsumsi dari offset paling awal. Penting Pada versi VVR sebelum 11.1, nilai consumer_group tidak didukung. Anda harus mengatur consumeFromCheckpoint ke true. Dalam kasus ini, log dikonsumsi dari offset yang dicatat dalam kelompok konsumen yang ditentukan, dan mode startup yang ditentukan di sini tidak berlaku.
  startTime	Waktu mulai konsumsi log.	String	Tidak	Waktu saat ini	Formatnya adalah yyyy-MM-dd hh:mm:ss. Parameter ini hanya berlaku ketika startupMode diatur ke timestamp. Catatan Parameter startTime dan stopTime didasarkan pada atribut __receive_time__ di SLS, bukan atribut __timestamp__.
  stopTime	Waktu berakhirnya konsumsi log.	String	Tidak	Tidak ada	Formatnya adalah yyyy-MM-dd hh:mm:ss. Catatan Parameter ini hanya digunakan untuk mengonsumsi log historis dan harus diatur ke titik waktu di masa lalu. Jika Anda mengaturnya ke waktu di masa depan, konsumsi mungkin berhenti lebih awal jika tidak ada log baru yang ditulis. Hal ini tampak sebagai gangguan aliran data tanpa pesan error. Jika Anda ingin program Flink keluar setelah konsumsi log selesai, Anda juga harus mengatur exitAfterFinish=true.
  consumerGroup	Nama grup konsumen.	String	Tidak	Tidak ada	Kelompok konsumen digunakan untuk mencatat progres konsumsi. Anda dapat menentukan nama kustom untuk kelompok konsumen. Formatnya tidak tetap. Catatan Beberapa pekerjaan tidak dapat menggunakan kelompok konsumen yang sama untuk konsumsi kolaboratif. Pekerjaan Flink yang berbeda harus memiliki kelompok konsumen yang berbeda. Jika pekerjaan Flink yang berbeda menggunakan kelompok konsumen yang sama, mereka akan mengonsumsi semua data. Hal ini karena ketika Flink mengonsumsi data dari SLS, alokasi partisi tidak dilakukan melalui kelompok konsumen SLS. Akibatnya, setiap konsumen mengonsumsi pesannya sendiri secara independen, meskipun kelompok konsumennya sama.
  consumeFromCheckpoint	Menentukan apakah akan mengonsumsi log dari checkpoint yang disimpan dalam kelompok konsumen yang ditentukan.	String	Tidak	false	true: Anda juga harus menentukan kelompok konsumen. Program Flink mengonsumsi log dari checkpoint yang disimpan dalam kelompok konsumen. Jika kelompok konsumen tidak memiliki checkpoint yang sesuai, log dikonsumsi dari waktu yang ditentukan oleh parameter startTime. false (default): Log tidak dikonsumsi dari checkpoint yang disimpan dalam kelompok konsumen yang ditentukan. Penting Parameter ini tidak lagi didukung mulai dari VVR 11.1. Untuk VVR 11.1 dan yang lebih baru, Anda harus mengatur startupMode ke consumer_group.
  maxRetries	Jumlah percobaan ulang setelah kegagalan membaca data dari SLS.	String	Tidak	3	Tidak ada.
  batchGetSize	Jumlah kelompok log yang dibaca dalam satu permintaan.	String	Tidak	100	Nilai batchGetSize tidak boleh melebihi 1.000. Jika tidak, akan muncul 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 Parameter ini sudah tidak digunakan lagi di VVR 11.3 dan yang lebih baru, tetapi tetap kompatibel dengan versi selanjutnya.	Pernyataan pra-pemrosesan untuk konsumsi data SLS.	String	Tidak	Tidak ada	Menggunakan parameter query memungkinkan Anda memfilter data sebelum dikonsumsi dari SLS. Hal ini mencegah semua data dikonsumsi ke Flink, sehingga menghemat biaya dan meningkatkan kecepatan pemrosesan. Misalnya, 'query' = '*| where request_method = ''GET''' menunjukkan bahwa sebelum Flink membaca data dari SLS, data tersebut dicocokkan dengan nilai bidang request_method yang bernilai 'get'. Catatan Parameter query memerlukan penggunaan Structured Process Language (SPL) untuk Simple Log Service. Untuk informasi selengkapnya, lihat Sintaksis SPL. Penting Parameter ini hanya didukung di Realtime Compute for Apache Flink VVR 8.0.1 atau yang lebih baru. Fitur ini dikenakan biaya oleh Simple Log Service. Untuk informasi selengkapnya, lihat Penagihan.
  processor	Prosesor Consum SLS. Jika parameter ini dan `query` dikonfigurasi bersamaan, `query` akan diutamakan dan `processor` tidak berlaku.	String	Tidak	Tidak ada	Menggunakan parameter processor memungkinkan Anda memfilter data sebelum dikonsumsi oleh Flink. Hal ini membantu menghemat biaya dan meningkatkan kecepatan pemrosesan. Kami menyarankan agar Anda menggunakan parameter processor daripada parameter query. Misalnya, 'processor' = 'test-filter-processor' menunjukkan bahwa data difilter oleh Prosesor Consum SLS sebelum dibaca oleh Flink. Catatan Prosesor harus menggunakan Structured Process Language (SPL) dari Simple Log Service (SLS). Untuk informasi selengkapnya, lihat Sintaksis SPL. Untuk membuat dan memperbarui Prosesor Consum SLS, lihat Kelola Prosesor Consum. Penting Parameter ini hanya didukung di Realtime Compute for Apache Flink VVR 11.3 atau yang lebih baru. Fitur ini dikenakan biaya oleh Simple Log Service. Untuk informasi selengkapnya, lihat Penagihan.

• Hanya untuk tabel sink

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Catatan
  topicField	Menentukan nama bidang. Nilai bidang ini akan menggantikan nilai bidang atribut __topic__ dan menunjukkan topik log.	String	Tidak	Tidak ada	Nilai parameter ini harus merupakan salah satu bidang yang ada dalam tabel.
  timeField	Menentukan nama bidang. Nilai bidang ini akan menggantikan nilai bidang atribut __timestamp__ dan menunjukkan waktu penulisan log.	String	Tidak	Waktu saat ini	Nilai parameter ini harus merupakan salah satu bidang yang ada dalam tabel, dan tipe bidangnya harus INT. Jika parameter ini tidak ditentukan, waktu saat ini digunakan secara default.
  sourceField	Menentukan nama bidang. Nilai bidang ini akan menggantikan nilai bidang atribut __source__ dan menunjukkan sumber log, seperti alamat IP mesin yang menghasilkan log.	String	Tidak	Tidak ada	Nilai parameter ini harus merupakan salah satu bidang yang ada dalam tabel.
  partitionField	Menentukan nama bidang. Saat data ditulis, nilai hash dihitung berdasarkan nilai kolom ini. Data dengan nilai hash yang sama ditulis ke shard yang sama.	String	Tidak	Tidak ada	Jika parameter ini tidak ditentukan, setiap data ditulis secara acak ke shard yang tersedia saat itu.
  buckets	Jumlah kelompok yang dikelompokkan ulang berdasarkan nilai hash saat partitionField ditentukan.	String	Tidak	64	Nilai parameter ini harus berada dalam rentang [1, 256] dan harus merupakan pangkat bulat dari 2. Jumlah bucket harus lebih besar dari atau sama dengan jumlah shard. Jika tidak, beberapa shard tidak akan menerima data.
  flushIntervalMs	Periode yang memicu penulisan data.	String	Tidak	2000	Satuan: milidetik.
  writeNullProperties	Menentukan apakah akan menulis nilai null sebagai string kosong ke SLS.	Boolean	Tidak	true	true (default): Nilai null ditulis sebagai string kosong ke log. false: Bidang dengan nilai null tidak ditulis ke log. Catatan Parameter ini hanya didukung di Realtime Compute for Apache Flink VVR 8.0.6 atau yang lebih baru.

Pemetaan tipe

Batasan

Fitur ini hanya didukung di Realtime Compute for Apache Flink VVR 11.1 atau yang lebih baru.

Sintaks

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

Item konfigurasi

Pemetaan tipe

Tabel berikut menunjukkan pemetaan tipe untuk data ingestion.

Inferensi skema tabel dan sinkronisasi perubahan

• Pre-konsumsi data shard dan inisialisasi skema tabel

  Konektor SLS mempertahankan skema Logstore yang sedang dibaca. Sebelum membaca data dari Logstore, konektor SLS mencoba melakukan pre-consume hingga maxPreFetchLogGroups kelompok log dari setiap shard. Konektor kemudian mengurai skema setiap log dalam kelompok log tersebut dan menggabungkan skema-skema tersebut untuk menginisialisasi skema tabel. Sebelum konsumsi data aktual dimulai, event pembuatan tabel yang sesuai dihasilkan berdasarkan skema yang diinisialisasi.

  Catatan

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

• Informasi primary key

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

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

• Inferensi skema dan sinkronisasi perubahan

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

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

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

  Konektor SLS menginferensi tipe semua bidang dalam setiap log sebagai String. Saat ini, hanya penambahan kolom yang didukung. Kolom baru ditambahkan di akhir skema saat ini dan diatur sebagai kolom nullable.