FETCH_CONTENT, TRY_FETCH_CONTENT - Realtime Compute for Apache Flink - Alibaba Cloud ドキュメントセンター

URI からファイルをダウンロードし、そのコンテンツをバイト配列として返します。

FETCH_CONTENT と TRY_FETCH_CONTENT は、指定された URI から非同期にファイルをダウンロードし、そのコンテンツを VARBINARY として返します。ダウンロードが失敗した場合の動作が異なります。

関数	ダウンロード成功時	リトライ回数の上限に達してもダウンロードに失敗した場合	適用シナリオ
`FETCH_CONTENT`	ファイルコンテンツを返します	例外をスローし、ジョブが失敗します	ファイルコンテンツが後続の処理に必須の入力であり、失敗したレコードをスキップできない場合。
`TRY_FETCH_CONTENT`	ファイルコンテンツを返します	NULL を返し、ジョブは他のレコードの処理を継続します	無効な URI、存在しないファイル、ネットワークエラーなど、失敗したレコードをスキップできる場合。

エンジン要件

FETCH_CONTENT 関数は、リアルタイムコンピューティングエンジン VVR 11.7.0 以降でのみサポートされます。
TRY_FETCH_CONTENT 関数は、リアルタイムコンピューティングエンジン VVR 11.9.0-preview.1 以降でのみサポートされます。

構文

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)

パラメーター

パラメーター

型

説明

uri

VARCHAR

ファイル URI です。サポートされているスキームには、HTTP、HTTPS、および Flink FileSystem でサポートされているすべてのスキームが含まれます。

http:// または https://：HTTP または HTTPS リンクです。
oss://：Alibaba Cloud OSS パスです。
hdfs://：HDFS パスです。
file://：ローカルファイルパスです。

concurrency

INTEGER

省略可能です。各 FETCH_CONTENT 関数インスタンスの専用 I/O スレッドプールのサイズです。プールは、ファイルシステムの読み取りと非同期 HTTP クライアントコールバックを処理します。省略した場合、既定値は max (8, 利用可能な JVM プロセッサ数) です。たとえば、4 プロセッサの場合は既定値が 8、16 プロセッサの場合は既定値が 16 になります。

説明

uri が NULL の場合は NULL を返します。uri が NULL でない場合、ダウンロードが失敗すると例外をスローします。
uri が OSS パスを指定している場合は、「Configure bucket authentication」に従ってアクセス認証情報を設定してください。

戻り値

型	説明
VARBINARY	ファイルコンテンツ

TRY_FETCH_CONTENT は、実行時のコンテンツ取得の失敗のみを許容します。引数の数や型が間違っている、同時実行数が範囲外であるなどの SQL 検証エラーは、ジョブの送信を失敗させます。

リトライとタイムアウトの設定

FETCH_CONTENT と TRY_FETCH_CONTENT は、非同期スカラー関数で共通のリトライおよびタイムアウト設定を使用します。リクエストの同時実行数、タイムアウト、およびリトライは、Flink Async Scalar オペレーターによって管理されます。SQL ジョブで SET ステートメントを使用して、関連パラメーターを変更できます。

次の例では、リトライ戦略を固定遅延リトライに設定し、最大リトライ回数を 2 に設定し、2 つのリクエスト間の待機時間を 1 s に設定し、単一レコードの非同期呼び出しタイムアウトを 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';

table.exec.async-scalar.max-attempts を 2 に設定すると、関数は最大 3 回 (最初のリクエストと 2 回のリトライ) のリクエストを送信します。すべてのリトライが完了する前にタイムアウトした場合、タイムアウトの結果が適用されます。FETCH_CONTENT は例外をスローし、TRY_FETCH_CONTENT は NULL を返します。

次の表は、すべてのパラメーターについて説明しています。

パラメーター	既定値	説明
table.exec.async-scalar.timeout	3 min	単一の入力レコードに対する非同期操作が完了するまでの最大待機時間です。
table.exec.async-scalar.max-attempts	3	最初の呼び出しを除く最大リトライ回数です。既定では、呼び出しは最大 4 回試行されます。
table.exec.async-scalar.max-concurrent-operations	10	サブタスクごとに許可される実行中の非同期操作の最大数です。`concurrency` と組み合わせることで、有効な同時実行数が決まります。このパラメーターは、サブタスクごとの実行中の入力を制限し、`concurrency` は、関数インスタンスごとの I/O エグゼキューターリソースを制限します。有効な同時実行数は、通常、この 2 つの値のうち小さい方の値に制限されます。
table.exec.async-scalar.retry-strategy	FIXED_DELAY	リトライ戦略です。有効な値： `FIXED_DELAY`：固定間隔でリトライします。 `NO_RETRY`：リトライしません。
table.exec.async-scalar.retry-delay	100 ms	各リトライ間の固定遅延です。

例1：ファイルコンテンツのダウンロード

テストデータ

表1. T1

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

テストクエリ

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

テスト結果

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

例2：ダウンロード失敗時にNULLを返す

テストデータ

表1. T1

id	uri (VARCHAR)
1	http://example.com/image_url
2	oss://example-bucket/example.pdf
3	invalid://path

TRY_FETCH_CONTENT を使用してファイルをダウンロードします。ダウンロードが失敗した場合、そのレコードの content 列には NULL が返され、ジョブが中断されることはありません。
```
SELECT
  id,
  TRY_FETCH_CONTENT(uri) AS content
FROM T1;
```

テスト結果

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

ダウンロードに失敗したレコードをフィルタリングして、後続のマルチモーダル推論関数に NULL を渡さないようにすることができます。
```
SELECT id, content
FROM (
  SELECT
    id,
    TRY_FETCH_CONTENT(uri) AS content
  FROM T1
)
WHERE content IS NOT NULL;
```

例3：同時実行数の指定

次の例では、各オペレーターインスタンスのコンテンツ取得の同時実行数を 8 に設定します。8 は、SQL ステートメント内で整数リテラルとして指定する必要があります。

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

次の呼び出しは無効であり、SQL 検証段階でエラーが発生します。

-- concurrency は列を参照できません。
SELECT TRY_FETCH_CONTENT(uri, concurrency_column) FROM T3;

-- concurrency の有効な値: 1 ～ 1024。
SELECT TRY_FETCH_CONTENT(uri, 0) FROM T3;
SELECT TRY_FETCH_CONTENT(uri, 1025) FROM T3;

Realtime Compute for Apache Flink:FETCH_CONTENT, TRY_FETCH_CONTENT