阿里云Flink SQL如何基于SPL实现弱结构化分析 - Simple Log Service

Anda dapat menggunakan Flink SQL bersama Simple Log Service Processing Language (SPL) untuk mengurai data log semi-terstruktur menjadi bidang terstruktur guna analisis, tanpa membuat penyimpanan log atau tabel temporary perantara.

Latar Belakang

Simple Log Service (SLS) adalah platform observabilitas dan analitik cloud-native yang menyediakan layanan real-time berskala besar dan berbiaya rendah untuk log, metrik, dan trace. Anda dapat mengumpulkan log sistem, log bisnis, dan data lainnya ke SLS untuk disimpan dan dianalisis. Realtime Compute for Apache Flink adalah platform analitik big data yang dibangun oleh Alibaba Cloud berdasarkan Apache Flink, yang banyak digunakan untuk analisis data real-time dan pemantauan risiko. Realtime Compute for Apache Flink secara native mendukung konektor SLS, sehingga memungkinkan Anda menggunakan SLS sebagai tabel sumber atau tabel hasil.

Konektor SLS untuk Realtime Compute for Apache Flink menangani log terstruktur secara langsung dengan memetakan satu per satu field log ke kolom tabel Flink SQL. Namun, banyak log bisnis tidak sepenuhnya terstruktur. Misalnya, seluruh konten log mungkin ditulis ke dalam satu field saja, sehingga memerlukan ekspresi reguler atau pemisahan berdasarkan delimiter untuk mengekstraksi field terstruktur. Topik ini menjelaskan cara menggunakan SPL dalam konektor SLS untuk menstrukturkan data tersebut, mencakup pembersihan log dan normalisasi format.

Data log semi-terstruktur

Log sampel berikut memiliki format kompleks yang mencampur string JSON dengan tipe data lainnya. Log ini berisi:

Field Payload, yang merupakan string JSON di mana field schedule bersarang di dalamnya juga berupa struktur JSON.
Field requestURL, yang merupakan path URL standar.
Field error, yang dimulai dengan string CouldNotExecuteQuery dan diikuti oleh struktur JSON.
Field __tag__:__path__, yang berisi path file log, di mana service_a kemungkinan merupakan nama layanan.
Field caller, yang berisi nama file dan nomor baris.

{
  "Payload": "{\"lastNotified\": 1705030483, \"serverUri\": \"http://test.alert.com/alert-api/tasks\", \"jobID\": \"44d6ce47bb4995ef0c8052a9a30ed6d8\", \"alertName\": \"alert-12345678-123456\", \"project\": \"test-sls-project\", \"projectId\": 123, \"aliuid\": \"1234567890\", \"alertDisplayName\": \"\\u6d4b\\u8bd5\\u963f\\u91cc\\u4e91\\u544a\\u8b66\", \"checkJobUri\": \"http://test.alert.com/alert-api/task_check\", \"schedule\": {\"timeZone\": \"\", \"delay\": 0, \"runImmediately\": false, \"type\": \"FixedRate\", \"interval\": \"1m\"}, \"jobRunID\": \"bf86aa5e67a6891d-61016da98c79b-5071a6b\", \"firedNotNotified\": 25161}",
  "TaskID": "bf86aa5e67a6891d-61016da98c79b-5071a6b-334f81a-5c38aaa1-9354-43ec-8369-4f41a7c23887",
  "TaskType": "ALERT",
  "__source__": "11.199.XXX.XXX",
  "__tag__:__hostname__": "iabcde12345.cloud.abc121",
  "__tag__:__path__": "/var/log/service_a.LOG",
  "caller": "executor/pool.go:64",
  "error": "CouldNotExecuteQuery : {\n    \"httpCode\": 404,\n    \"errorCode\": \"LogStoreNotExist\",\n    \"errorMessage\": \"logstore k8s-event does not exist\",\n    \"requestID\": \"65B7C10AB43D9895A8C3DB6A\"\n}",
  "requestURL": "/apis/autoscaling/v2beta1/namespaces/python-etl/horizontalpodautoscalers/cn-shenzhen-56492-1234567890123?timeout=30s",
  "ts": "2024-01-29 22:57:13"
}

Kebutuhan strukturisasi data

Untuk mengekstraksi informasi bernilai dari log ini, Anda harus mentransformasi data dengan mengekstraksi field kunci untuk analisis.

Dari field error, ekstrak httpCode, errorCode, errorMessage, dan requestID.
Dari field __tag__:__path_, ekstrak _service_a sebagai serviceName.
Dari field caller, ekstrak pool.go sebagai fileName dan 64 sebagai fileNo.
Dari field Payload, ekstrak project. Dari objek schedule bersarang dalam Payload, ekstrak type dan beri nama scheduleType.
Ubah nama field __source__ menjadi serviceIP.

Semua field lainnya dibuang. Daftar akhir field yang diperlukan adalah: httpCode, errorCode, errorMessage, requestID, serviceName, fileName, fileNo, project, scheduleType, dan serviceIP.

Solusi

Terdapat beberapa metode untuk mentransformasi data. Bagian ini membandingkan solusi yang menggunakan SLS dan Flink, masing-masing cocok untuk skenario berbeda.

Solusi transformasi data: Di Konsol SLS, buat pekerjaan transformasi data untuk membersihkan data dan menyimpannya di penyimpanan log target.
Solusi Flink: Definisikan error dan payload sebagai field di tabel sumber. Gunakan fungsi ekspresi reguler dan JSON Flink SQL untuk mengurai field-field tersebut, tulis hasilnya ke tabel temporary, lalu lakukan analisis pada tabel tersebut.
Solusi SPL: Konfigurasikan pernyataan SPL di konektor Flink SLS untuk mentransformasi data. Tabel sumber Flink kemudian didefinisikan dengan skema terstruktur akhir.

Mengonfigurasi SPL di konektor SLS merupakan pendekatan yang lebih ringan. Untuk data log semi-terstruktur, solusi SPL menghindari pembuatan penyimpanan log perantara (yang diperlukan oleh solusi transformasi data) dan menghindari pembuatan tabel temporary di Flink (yang diperlukan oleh solusi Flink). Dengan melakukan transformasi data lebih dekat ke sumber, Anda dapat fokus pada logika bisnis di platform komputasi, sehingga menciptakan pemisahan tanggung jawab yang lebih jelas.

Gunakan SPL di Flink

1. Siapkan data di SLS

Pastikan Anda telah mengaktifkan Simple Log Service dan membuat proyek dan penyimpanan log.
Tulis cuplikan log di atas ke penyimpanan log target Anda menggunakan SDK untuk mensimulasikan data sampel.

Setelah data ditulis, buka tab Raw Logs di Konsol Simple Log Service untuk melihat data simulasi. Konten log mencakup field seperti TaskType (dengan nilai ALERT), Payload (yang berisi alertName, project, dan aliuid), serta informasi error (kode error LogStoreNotExist, pesan logstore k8s-event does not exist, dan status HTTP 404).
Di penyimpanan log, tulis pernyataan pipeline SPL SLS dan pratinjau hasilnya.

Setelah Anda menjalankan pernyataan SPL, tab Raw Logs menampilkan catatan log yang telah diurai, yang mencakup field terstruktur seperti errorCode, errorMessage, fileName, fileNo, httpCode, project, requestID, scheduleType, dan serviceHost.

Pernyataan kueri SPL adalah sebagai berikut. Sintaksis pipeline SPL menggunakan delimiter pipe (|) untuk memisahkan perintah. Anda dapat memasukkan satu perintah dalam satu waktu untuk melihat hasil langsungnya, lalu tambahkan lebih banyak pipe untuk membangun kueri akhir secara iteratif. Untuk informasi lebih lanjut, lihat Sintaksis kueri scan.
```
* | project Payload, error, "__tag__:__path__", "__tag__:__hostname__", caller 
 | parse-json Payload 
 | project-away Payload 
 | parse-regexp error, 'CouldNotExecuteQuery : ({[\w":\s,\-}]+)' as errorJson 
 | parse-json errorJson 
 | parse-regexp "__tag__:__path__", '\/var\/log\/([\w\_]+).LOG' as serviceName 
 | parse-regexp caller, '\w+/([\w\.]+):(\d+)' as fileName, fileNo 
 | project-rename serviceHost="__tag__:__hostname__" 
 | extend scheduleType = json_extract_scalar(schedule, '$.type') 
 | project httpCode, errorCode,errorMessage,requestID,fileName, fileNo, serviceHost,scheduleType, project
```
Penjelasan sintaksis:
- Baris 1: Perintah project menyimpan field Payload, error, __tag__:__path__, dan caller untuk penguraian, serta membuang semua field lainnya.
- Baris 2: Perintah parse-json mengembangkan string Payload menjadi objek JSON. Field tingkat atasnya, seperti lastNotified, serviceUri, dan jobID, ditambahkan ke hasil.
- Baris 3: Perintah project-away menghapus field Payload asli.
- Baris 4: Perintah parse-regexp menggunakan ekspresi reguler untuk mengekstraksi bagian JSON dari field error dan menetapkannya ke field baru bernama errorJson.
- Baris 5: Perintah parse-json mengembangkan field errorJson, mengekstraksi httpCode, errorCode, dan errorMessage.
- Baris 6: Perintah parse-regexp menggunakan ekspresi reguler untuk mengekstraksi nama file dari __tag__:__path__ dan memberinya nama serviceName.
- Baris 7: Perintah parse-regexp menggunakan ekspresi reguler untuk mengekstraksi nama file dan nomor baris dari field caller dan menetapkannya ke field fileName dan fileNo.
- Baris 8: Perintah project-rename mengubah nama field __tag__:__hostname__ menjadi serviceHost.
- Baris 9: Perintah extend menggunakan fungsi json_extract_scalar untuk mengekstraksi field type dari objek schedule dan memberinya nama scheduleType.
- Baris 10: Perintah project hanya menyimpan field akhir yang diperlukan, termasuk field project yang diekstraksi dari Payload.

2. Buat pekerjaan SQL

Login ke Konsol Realtime Compute for Apache Flink dan klik ruang kerja target.
Di panel navigasi kiri, pilih Data Development > ETL.
Klik Create. Di kotak dialog New Draft, pilih SQL Scripts > Blank Stream Draft, lalu klik Next.

Di editor draft, masukkan pernyataan berikut untuk membuat tabel temporary.

CREATE TEMPORARY TABLE sls_input_complex (
  errorCode STRING,
  errorMessage STRING,
  fileName STRING,
  fileNo STRING,
  httpCode STRING,
  requestID STRING,
  scheduleType STRING,
  serviceHost STRING,
  project STRING,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-beijing-intranet.log.aliyuncs.com',
  'accessId' = '${yourAccessKeyID}',
  'accessKey' = '${yourAccessKeySecret}',
  'starttime' = '2024-02-01 10:30:00',
  'project' ='${project}',
  'logstore' ='${logtore}',
  'query' = '* | project Payload, error, "__tag__:__path__", "__tag__:__hostname__", caller | parse-json Payload | project-away Payload | parse-regexp error, ''CouldNotExecuteQuery : ({[\w":\s,\-}]+)'' as errorJson | parse-json errorJson | parse-regexp "__tag__:__path__", ''\/var\/log\/([\w\_]+).LOG'' as serviceName | parse-regexp caller, ''\w+/([\w\.]+):(\d+)'' as fileName, fileNo | project-rename serviceHost="__tag__:__hostname__" | extend scheduleType = json_extract_scalar(schedule, ''$.type'') | project httpCode, errorCode,errorMessage,requestID,fileName, fileNo, serviceHost,scheduleType,project'
  );

Tabel berikut menjelaskan parameter dalam klausa WITH. Ganti nilai contoh dengan nilai aktual Anda.

Parameter	Deskripsi	Contoh
connector	Jenis konektor. Konektor yang didukung.	sls
endpoint	Titik akhir internal SLS. Titik akhir.	cn-hangzhou-intranet.log.aliyuncs.com
accessId	ID AccessKey Anda. Buat AccessKey.	LTAI****************
accessKey	Rahasia AccessKey. Buat AccessKey.	yourAccessKeySecret
starttime	Waktu mulai konsumsi log.	2025-02-19 00:00:00
project	Nama proyek SLS.	test-project
logstore	Nama Logstore SLS.	clb-access-log
query	Pernyataan SPL. Escape literal string dengan dua tanda petik tunggal (`''`) di Flink SQL.	* \| where slbid = ''slb-01''

Pilih pernyataan SQL, klik kanan, lalu pilih Running untuk terhubung ke Simple Log Service.

CREATE TEMPORARY TABLE sls_input_complex (
  errorCode STRING,
  errorMessage STRING,
  fileName STRING,
  fileNo STRING,
  httpCode STRING,
  requestID STRING,
  scheduleType STRING,
  serviceHost STRING,
  project STRING,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' = 'cn-hxxx',
  'accessId' = 'xxx',
  'accessKey' = 'xxx',
  'starttime' = 'xxx',
  'project' = 'xxx',
  'logstore' = 'clb7xxx',
  'query' = '* | prxxx", "__tag__:__hosxxx'
);

3. Jalankan kueri dan lihat hasilnya

Di editor pekerjaan, masukkan pernyataan berikut untuk mengkueri data:
```
SELECT * FROM sls_input_complex;
```
Klik Debug di pojok kanan atas. Di kotak dialog, pilih Create new session cluster dari daftar drop-down Session Cluster dan konfigurasikan sebagai berikut.

Tetapkan Name menjadi demo-test, Deployment Target menjadi default-queue, Status menjadi RUNNING, dan Engine Version menjadi vvr-8.0.11-flink-1.17, lalu klik Create Session Cluster.
Di kotak dialog debug, pilih session cluster yang baru saja Anda buat, lalu klik OK.

Catatan: Debugging dengan tabel sumber SLS akan memajukan offset kelompok konsumen. Pekerjaan yang dideploy akan dilanjutkan dari offset baru ini.
Di tab Results, Anda dapat melihat bahwa setiap kolom dalam tabel sesuai dengan field yang diproses oleh kueri SPL.

Setelah Anda Debug kueri, tabel hasil menampilkan field yang telah diurai. Sebagai contoh, kolom errorCode berisi LogStoreNotExist, kolom errorMessage berisi logstore k8s-event does not exist, dan kolom httpCode berisi 404.