Cara menerapkan pemfilteran baris dan pemangkasan kolom di Alibaba Cloud Flink SQL menggunakan SPL - Simple Log Service

SPL memungkinkan pemfilteran baris dan pemangkasan kolom di server Simple Log Service (SLS) sebelum data mencapai Flink, sehingga mengurangi beban jaringan dan biaya komputasi.

Latar Belakang

Saat Anda menggunakan SLS sebagai tabel sumber di Realtime Compute for Apache Flink, konektor mengonsumsi semua catatan Logstore setelah waktu mulai tertentu. Hal ini menimbulkan dua masalah:

Konektor menarik terlalu banyak baris atau kolom dari sumber, sehingga meningkatkan beban jaringan.
Memfilter data yang tidak diperlukan di Flink membuang sumber daya komputasi.

SPL mengatasi hal ini melalui filter pushdown dan projection pushdown untuk konektor SLS. Konfigurasikan parameter query untuk mendorong kondisi filter dan bidang yang diproyeksikan ke server, sehingga menghindari transfer dan komputasi dataset lengkap.

Cara kerja

Tanpa pernyataan SPL: Flink menarik semua baris dan kolom dari SLS untuk komputasi.
Dengan pernyataan SPL: Jika pernyataan SPL mencakup pemfilteran baris atau pemangkasan kolom, Flink hanya menarik subset yang sesuai.

Prasyarat

Anda telah mengaktifkan Simple Log Service dan membuat Proyek dan Logstore.

Contoh ini menggunakan Logstore yang diisi dengan log SLB Layer-7 simulasi. Data berisi lebih dari 10 bidang dan dihasilkan secara terus-menerus. Contoh entri log:

{
  "__source__": "127.0.0.1",
  "__tag__:__receive_time__": "1706531737",
  "__time__": "1706531727",
  "__topic__": "slb_layer7",
  "body_bytes_sent": "3577",
  "client_ip": "114.137.XXX.XXX",
  "host": "www.pi.mock.com",
  "http_host": "www.cwj.mock.com",
  "http_user_agent": "Mozilla/5.0 (Windows NT 6.2; rv:22.0) Gecko/20130405 Firefox/23.0",
  "request_length": "1662",
  "request_method": "GET",
  "request_time": "31",
  "request_uri": "/request/path-0/file-3",
  "scheme": "https",
  "slbid": "slb-02",
  "status": "200",
  "upstream_addr": "42.63.XXX.XXX",
  "upstream_response_time": "32",
  "upstream_status": "200",
  "vip_addr": "223.18.XX.XXX"
}

Bidang slbid memiliki tiga nilai berbeda. Kueri selama 15 menit menunjukkan bahwa slb-01 dan slb-02 memiliki jumlah yang sebanding.

Jalankan * | select slbid, count(1) cnt group by slbid di editor kueri. Hasilnya: slb-02 memiliki 2.001 entri, slb-01 memiliki 1.986 entri, dan lb-uf6qfcvsrouodhvw39oog memiliki 4.144 entri.

Prosedur

Pemfilteran baris: Dorong kondisi filter ke server SLS melalui parameter query, sehingga mencegah transfer dataset lengkap.

Pemangkasan kolom: Dorong bidang yang diproyeksikan ke server SLS melalui parameter query, sehingga hanya mengembalikan kolom yang ditentukan.

Pemfilteran baris

Langkah 1: Buat pekerjaan SQL

Masuk 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.

Salin SQL tabel temporary berikut ke editor.

CREATE TEMPORARY TABLE sls_input(
  request_uri STRING,
  scheme STRING,
  slbid STRING,
  status STRING,
  `__topic__` STRING METADATA VIRTUAL,
  `__source__` STRING METADATA VIRTUAL,
  `__timestamp__` STRING METADATA VIRTUAL,
   __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou-intranet.log.aliyuncs.com',
  'accessId' = 'yourAccessKeyID',
  'accessKey' = 'yourAccessKeySecret',
  'starttime' = '2025-02-19 00:00:00',
  'project' ='test-project',
  'logstore' ='clb-access-log',
  'query' = '* | where slbid = ''slb-01'''
);

Parameter:

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. Gunakan tanda petik tunggal ganda (`''`) untuk meng-escape literal string di Flink SQL.	* \| where slbid = ''slb-01''

Pilih pernyataan SQL tersebut, klik kanan, lalu klik Running untuk terhubung ke SLS.

CREATE TEMPORARY TABLE sls_input(
  request_uri STRING,
  scheme STRING,
  slbid STRING,
  status STRING,
  `__topic__` STRING METADATA VIRTUAL,
  `__source__` STRING METADATA VIRTUAL,
  `__timestamp__` STRING METADATA VIRTUAL,
  __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' = 'cn-haxxx',
  'accessId' = 'xxx',
  'accessKey' = 'xxx',
  'starttime' = '202xxx',
  'project' = 'test-pxxx',
  'logstore' = 'clb-axxx',
  'query' = '*|whexxx'
);

Langkah 2: Jalankan kueri dan lihat hasilnya

Jalankan kueri agregat berikut pada bidang slbid.

SELECT slbid, count(1) as slb_cnt FROM sls_input GROUP BY slbid;

Klik Debug di pojok kanan atas. Di kotak dialog, pilih Create new session cluster dari daftar drop-down Session Cluster dan konfigurasikan sebagai berikut.

Atur 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.
Hasil menunjukkan bahwa slbid selalu slb-01, yang mengonfirmasi bahwa sls_input hanya berisi baris yang sesuai dengan kondisi filter.

Nilai slb_cnt adalah 185.

Pemangkasan kolom

Langkah 1: Buat pekerjaan SQL

Masuk 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.

Salin SQL tabel temporary berikut ke editor. Dibandingkan dengan pemfilteran baris, parameter query menambahkan pernyataan proyeksi. Perintah dirangkai dengan | (seperti pipa Unix), dan hanya bidang yang diproyeksikan yang diambil dari server SLS.

CREATE TEMPORARY TABLE sls_input_project(
  request_uri STRING,
  scheme STRING,
  slbid STRING,
  status STRING,
  `__topic__` STRING METADATA VIRTUAL,
  `__source__` STRING METADATA VIRTUAL,
  `__timestamp__` STRING METADATA VIRTUAL,
   __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou-intranet.log.aliyuncs.com',
  'accessId' = 'yourAccessKeyID',
  'accessKey' = 'yourAccessKeySecret',
  'starttime' = '2025-02-19 00:00:00',
  'project' ='test-project',
  'logstore' ='clb-access-log',
  'query' = '* | where slbid = ''slb-01'' | project request_uri, scheme, slbid, status, __topic__, __source__, "__tag__:__receive_time__"'
);

Parameter:

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. Gunakan tanda petik tunggal ganda (`''`) untuk meng-escape literal string di Flink SQL.	* \| where slbid = ''slb-01''

Pilih pernyataan SQL tersebut, klik kanan, lalu klik Running untuk terhubung ke SLS.

Langkah 2: Jalankan kueri dan lihat hasilnya

Jalankan kueri agregat berikut pada bidang slbid.

SELECT slbid, count(1) as slb_cnt FROM sls_input_project GROUP BY slbid;

Klik Debug di pojok kanan atas. Di kotak dialog, pilih Create new session cluster dari daftar drop-down Session Cluster dan konfigurasikan sebagai berikut.

Atur 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.
Hasilnya mirip dengan kasus penggunaan pemfilteran baris.

Catatan
Berbeda dengan pemfilteran baris, pemangkasan kolom hanya mengembalikan bidang yang ditentukan, sehingga lebih lanjut mengurangi lalu lintas jaringan.

Nilai slb_cnt adalah 185.