FETCH_CONTENT, TRY_FETCH_CONTENT - Realtime Compute for Apache Flink

Mengunduh file dari URI dan mengembalikan isinya sebagai larik byte.

FETCH_CONTENT dan TRY_FETCH_CONTENT mengunduh file secara asinkron dari URI yang ditentukan dan mengembalikan isinya sebagai VARBINARY. Keduanya berbeda dalam perilaku saat pengunduhan gagal:

Fungsi	Pengunduhan berhasil	Pengunduhan gagal setelah semua percobaan ulang habis	Skenario penerapan
`FETCH_CONTENT`	Mengembalikan isi file	Melemparkan exception dan pekerjaan gagal	Isi file merupakan input wajib untuk pemrosesan lanjutan. Catatan yang gagal tidak boleh dilewati.
`TRY_FETCH_CONTENT`	Mengembalikan isi file	Mengembalikan NULL dan pekerjaan melanjutkan pemrosesan catatan lainnya	Catatan yang gagal dapat dilewati, seperti URI tidak valid, file tidak ada, dan error jaringan.

Persyaratan engine

Hanya engine Realtime Compute VVR 11.7.0 dan versi lebih baru yang mendukung fungsi FETCH_CONTENT.
Hanya engine Realtime Compute VVR 11.9.0-preview.1 dan versi lebih baru yang mendukung fungsi TRY_FETCH_CONTENT.

Sintaks

VARBINARY FETCH_CONTENT(VARCHAR uri)
VARBINARY FETCH_CONTENT(VARCHAR uri, INTEGER concurrency)

VARBINARY TRY_FETCH_CONTENT(VARCHAR uri)
VARBINARY TRY_FETCH_CONTENT(VARCHAR uri, INTEGER concurrency)

Parameter

Parameter

Type

Deskripsi

uri

VARCHAR

URI file. Skema yang didukung mencakup HTTP, HTTPS, dan skema apa pun yang didukung oleh Flink FileSystem:

http:// atau https://: Tautan HTTP atau HTTPS.
oss://: Jalur Alibaba Cloud OSS.
hdfs://: Jalur HDFS.
file://: Jalur file lokal.

concurrency

INTEGER

Opsional. Ukuran kolam thread I/O khusus untuk setiap instans fungsi FETCH_CONTENT. Kolam ini melayani pembacaan sistem file dan callback klien HTTP asinkron. Jika tidak ditentukan, nilai default-nya adalah max(8, jumlah prosesor JVM yang tersedia). Misalnya, dengan 4 prosesor nilai default-nya adalah 8; dengan 16 prosesor nilai default-nya adalah 16.

Catatan

Mengembalikan NULL jika uri bernilai NULL. Untuk uri yang tidak NULL, pengunduhan yang gagal akan melemparkan exception.
Jika uri menentukan jalur OSS, konfigurasikan kredensial akses sebagaimana dijelaskan dalam Configure bucket authentication.

Nilai kembalian

Type	Deskripsi
VARBINARY	Isi file

TRY_FETCH_CONTENT hanya mentoleransi kegagalan pengambilan konten pada waktu proses. Error validasi SQL, seperti jumlah atau tipe argumen yang salah atau nilai concurrency di luar rentang, tetap menyebabkan pengiriman pekerjaan gagal.

Konfigurasi retry dan timeout

FETCH_CONTENT dan TRY_FETCH_CONTENT menggunakan konfigurasi retry dan timeout terpadu untuk fungsi skalar asinkron. Konkurensi permintaan, timeout, dan retry dikelola oleh operator Flink Async Scalar. Anda dapat mengubah parameter terkait dengan pernyataan SET dalam pekerjaan SQL Anda.

Contoh berikut mengatur strategi retry menjadi retry dengan penundaan tetap dengan maksimum 2 kali percobaan ulang, menunggu 1 s antara dua permintaan, dan mengatur timeout pemanggilan asinkron untuk satu catatan menjadi 30 s.

SET 'table.exec.async-scalar.retry-strategy' = 'FIXED_DELAY';
SET 'table.exec.async-scalar.max-attempts' = '2';
SET 'table.exec.async-scalar.retry-delay' = '1 s';
SET 'table.exec.async-scalar.timeout' = '30 s';

Ketika table.exec.async-scalar.max-attempts diatur ke 2, fungsi mengirim hingga 3 permintaan: permintaan pertama ditambah 2 kali percobaan ulang. Jika timeout tidak memberikan cukup waktu untuk semua percobaan ulang, hasil timeout berlaku: FETCH_CONTENT melemparkan exception dan TRY_FETCH_CONTENT mengembalikan NULL.

Tabel berikut menjelaskan semua parameter.

Parameter	Nilai default	Deskripsi
table.exec.async-scalar.timeout	3 min	Waktu maksimum untuk menunggu operasi asinkron selesai untuk satu catatan input.
table.exec.async-scalar.max-attempts	3	Maksimum percobaan ulang tidak termasuk panggilan pertama. Secara default, panggilan dilakukan hingga 4 kali.
table.exec.async-scalar.max-concurrent-operations	10	Jumlah maksimum operasi asinkron yang sedang berjalan yang diizinkan per subtask. Digabungkan dengan `concurrency`, parameter ini menentukan konkurensi efektif: parameter ini membatasi jumlah input yang sedang diproses per subtask, sedangkan `concurrency` membatasi sumber daya pelaksana I/O per instans fungsi. Konkurensi efektif biasanya dibatasi oleh nilai yang lebih kecil dari keduanya.
table.exec.async-scalar.retry-strategy	FIXED_DELAY	Strategi retry. Nilai yang valid: `FIXED_DELAY`: retry dengan interval tetap. `NO_RETRY`: tidak melakukan retry.
table.exec.async-scalar.retry-delay	100 ms	Penundaan tetap antara dua percobaan retry.

Contoh 1: Mengunduh isi file

Data uji

Tabel 1. T1

input

uri (VARCHAR)

1

http://example.com/image_url

2

oss://example-bucket/example.pdf

3

NULL

Kueri uji

SELECT 
    id,
    FETCH_CONTENT(uri) AS `value`
FROM 
    T1;

Hasil uji

id (INT)

value (VARBINARY)

1

x'ffd8ffe00010......'

2

x'aaffd8ffe000......'

3

NULL

Contoh 2: Mengembalikan NULL saat pengunduhan gagal

Data uji

Tabel 1. T1

input

uri (VARCHAR)

1

http://example.com/image_url

2

oss://example-bucket/example.pdf

3

invalid://path
Gunakan TRY_FETCH_CONTENT untuk mengunduh file. Jika pengunduhan gagal, content dari catatan tersebut mengembalikan NULL, dan pekerjaan tidak terganggu oleh kegagalan tersebut.
```
SELECT
  id,
  TRY_FETCH_CONTENT(uri) AS content
FROM T2;
```
Hasil uji

id (INT)

value (VARBINARY)

1

x'ffd8ffe00010......'

2

NULL

3

NULL
Anda dapat memfilter catatan yang gagal diunduh untuk menghindari pengiriman NULL ke fungsi inferensi multimodal di hilir.
```
SELECT id, content
FROM (
  SELECT
    id,
    TRY_FETCH_CONTENT(uri) AS content
  FROM T2
)
WHERE content IS NOT NULL;
```

Contoh 3: Menentukan concurrency

Contoh berikut mengatur concurrency pengambilan konten untuk setiap instans operator menjadi 8. Nilai 8 harus berupa literal integer dalam pernyataan SQL.

SELECT
  id,
  TRY_FETCH_CONTENT(uri, 8) AS content
FROM T2;

Pemanggilan berikut tidak valid dan menyebabkan error pada tahap validasi SQL.

-- concurrency tidak dapat mereferensikan kolom.
SELECT TRY_FETCH_CONTENT(uri, concurrency_column) FROM T3;

-- Nilai concurrency yang valid: 1 hingga 1024.
SELECT TRY_FETCH_CONTENT(uri, 0) FROM T3;
SELECT TRY_FETCH_CONTENT(uri, 1025) FROM T3;

Referensi

Fungsi bawaan: Supported functions.
Fungsi yang ditentukan pengguna: UDFs dan Manage user-defined functions (UDFs).

input	uri (VARCHAR)
1	http://example.com/image_url
2	oss://example-bucket/example.pdf
3	NULL

id (INT)	value (VARBINARY)
1	x'ffd8ffe00010......'
2	x'aaffd8ffe000......'
3	NULL