Elasticsearch コネクタの使用方法について、制限、構文、パラメータ、例などを含めて説明します。 - Realtime Compute for Apache Flink

このトピックでは、Elasticsearch コネクターの使用方法を説明します。

背景

Alibaba Cloud Elasticsearch は、オープンソースの Elasticsearch と互換性があり、データ分析およびデータ検索用の Security、Machine Learning、Graph、APM などの商用機能を備えています。また、アクセス制御、セキュリティ監視とアラート、自動レポート生成など、エンタープライズグレードのサービスを提供します。

次の表に、Elasticsearch コネクタの機能を示します。

項目	説明
テーブルタイプ	ソーステーブル、ディメンションテーブル、シンクテーブル
実行モード	バッチモードとストリーミングモード
データフォーマット	JSON
メトリック	メトリックソーステーブル pendingRecords numRecordsIn numRecordsInPerSecond numBytesIn numBytesInPerSecond ディメンションテーブルなしシンクテーブル (Ververica Runtime (VVR) 6.0.6 以降) numRecordsOut numRecordsOutPerSecond 説明これらのメトリックの詳細については、「Metrics」をご参照ください。
API タイプ	DataStream API と SQL
シンクテーブルでのデータの更新または削除	対応しています

前提条件

Elasticsearch インデックスを作成済みであること。詳細については、「クイックスタート」をご参照ください。
Elasticsearch インスタンスに、パブリックまたはプライベート IP アドレスホワイトリストを設定済みであること。詳細については、「IP アドレスホワイトリストの管理」をご参照ください。

制限事項

ソーステーブルおよびディメンションテーブルは、Elasticsearch 6.8.x 以降に対応しています。

説明
ソーステーブルおよびディメンションテーブルで Elasticsearch 8.x を使用するには、VVR 11.6 以降が必要です。
シンクテーブルは、Elasticsearch 6.x、7.x、および 8.x のみに対応しています。
フル Elasticsearch ソーステーブルにのみ対応しており、インクリメンタルソーステーブルには対応していません。

構文

ソーステーブル

Elasticsearch 8.x

CREATE TABLE elasticsearch_source(
  name STRING,
  location STRING,
  value FLOAT
) WITH (
  'connector' ='elasticsearch-8',
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>'
);

その他のバージョン

CREATE TABLE elasticsearch_source(
  name STRING,
  location STRING,
  value FLOAT
) WITH (
  'connector' ='elasticsearch',
  'endPoint' = '<yourEndPoint>',
  'indexName' = '<yourIndexName>'
);

ディメンションテーブル
Elasticsearch 8.x
```
CREATE TABLE es_dim(
  field1 STRING, -- JOIN キーとして使用する場合は、STRING 型である必要があります。
  field2 FLOAT,
  field3 BIGINT,
  PRIMARY KEY (field1) NOT ENFORCED
) WITH (
  'connector' ='elasticsearch-8',
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>'
);
```
その他のバージョン
```
CREATE TABLE es_dim(
  field1 STRING, -- JOIN キーとして使用する場合は、STRING 型である必要があります。
  field2 FLOAT,
  field3 BIGINT,
  PRIMARY KEY (field1) NOT ENFORCED
) WITH (
  'connector' ='elasticsearch',
  'endPoint' = '<yourEndPoint>',
  'indexName' = '<yourIndexName>'
);
```
説明
- プライマリキーが指定されている場合、JOIN キーとして使用できるフィールドは 1 つのみであり、そのフィールドの値が対応する Elasticsearch インデックスのドキュメント ID として使用される必要があります。
- プライマリキーが指定されていない場合、1 つ以上のフィールドを JOIN キーとして使用できます。これらのキーは、対応する Elasticsearch ドキュメント内のフィールドである必要があります。
- STRING フィールドの場合、コネクタは互換性のため、デフォルトでフィールド名に .keyword サフィックスを追加します。これにより Elasticsearch の TEXT フィールドと一致しない場合は、ignoreKeywordSuffix オプションを true に設定してください。
シンクテーブル
```
CREATE TABLE es_sink(
  user_id   STRING,
  user_name   STRING,
  uv BIGINT,
  pv BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'elasticsearch-7', -- Elasticsearch 6.x を使用する場合は、'elasticsearch-6' に設定してください。
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>'
);
```
説明
- Elasticsearch シンクテーブルは、プライマリキーが定義されているかどうかに応じて、upsert モード または append モード で動作します。
  - プライマリキーが定義されている場合、その値がドキュメント ID として使用されます。シンクテーブルは upsert モード で動作し、UPDATE および DELETE 操作を処理できます。
  - プライマリキーが定義されていない場合、Elasticsearch はランダムなドキュメント ID を自動的に生成します。シンクテーブルは append モード で動作し、INSERT メッセージのみを処理できます。
- BYTES、ROW、ARRAY、MAPなどのデータ型には、対応する文字列表現がありません。そのため、これらのデータ型のフィールドをプライマリキーとして使用することはできません。
- DDL 内のフィールドは、Elasticsearch ドキュメント内のフィールドに対応します。ドキュメント ID などのメタデータはシンクテーブルに書き込むことはできません。これらのメタデータは Elasticsearch クラスタによって管理されます。

WITH オプション

ソーステーブル

パラメーター	説明	タイプ	必須	デフォルト	備考
connector	ソーステーブルのタイプです。	文字列	はい	なし	有効な値： `elasticsearch` または `elasticsearch-8`。説明 VVR 11.6 以降でのみ `elasticsearch-8` の値がサポートされています。
endPoint	Elasticsearch クラスターのサーバーアドレスです。	文字列	はい	なし	レガシーオプション名。
hosts	Elasticsearch クラスターのサーバーアドレスです。	文字列	はい	なし	`elasticsearch-8` で使用します。
indexName	インデックスの名前です。	文字列	はい	なし	レガシーオプション名。
index	インデックスの名前です。	文字列	はい	なし	`elasticsearch-8` で使用します。
accessId	認証用のユーザー名です。	文字列	いいえ	なし	デフォルトでは、このパラメーターは空で、認証は行われません。accessId を指定する場合は、空でない accessKey を指定する必要があります。説明 `elasticsearch-8` では、`username` および `password` パラメーターを使用します。重要ユーザー名とパスワードの漏洩を防ぐために、プロジェクト変数を使用することを推奨します。詳細については、「プロジェクト変数」をご参照ください。
username	認証用のユーザー名です。	文字列	いいえ	なし
accessKey	認証用のパスワードです。	文字列	いいえ	なし
password	認証用のパスワードです。	文字列	いいえ	なし
typeNames	タイプの名前です。	文字列	いいえ	_doc	Elasticsearch 7.0 以降では、このオプションを設定しないことを推奨します。
batchSize	スクロールリクエストごとに Elasticsearch クラスターから取得するドキュメントの最大数です。	Int	いいえ	2000	なし
keepScrollAliveSecs	スクロールコンテキストを維持する最大時間です。	Int	いいえ	3600	単位：秒。

シンクテーブル

パラメーター	説明	タイプ	必須	デフォルト	備考
connector	シンクテーブルのタイプです。	文字列	はい	なし	値は `elasticsearch-6`、`elasticsearch-7`、または `elasticsearch-8` である必要があります。説明 `elasticsearch-8` の値をサポートしているのは、VVR 8.0.5 以降のみです。
hosts	Elasticsearch クラスターのサーバーアドレスです。	文字列	はい	なし	例： `127.0.0.1:XXXX`。
index	インデックス名です。	文字列	はい	なし	シンクテーブルは、静的インデックスと動的インデックスの両方をサポートしています：静的インデックスの場合、値は `myusers` などのプレーン文字列である必要があります。すべてのレコードは `myusers` インデックスに書き込まれます。動的インデックスの場合、`{field_name}` を使用してレコードからフィールド値を参照し、ターゲットインデックスを動的に生成できます。また、`{field_name\|date_format_string}` を使用して、TIMESTAMP、DATE、TIME データ型のフィールド値を、`date_format_string` で指定した形式に変換できます。`date_format_string` は、Java の DateTimeFormatter と互換性があります。たとえば、インデックスを `myusers-{log_ts\|yyyy-MM-dd}` に設定した場合、log_ts フィールド値が `2020-03-27 12:25:55` のレコードは、`myusers-2020-03-27` インデックスに書き込まれます。
document-type	ドキュメントタイプです。	文字列	`elasticsearch-6`：はい `elasticsearch-7`：サポートされていません	なし	コネクタタイプが `elasticsearch-6` の場合、このパラメーターの値は Elasticsearch の type パラメーターの値と一致している必要があります。
username	認証に使用するユーザー名です。	文字列	いいえ	なし	デフォルトでは、認証は無効です。`username` を指定する場合は、空でない `password` も指定する必要があります。重要ユーザー名とパスワードの漏洩を防ぐために、プロジェクト変数を使用することを推奨します。詳細については、「プロジェクト変数」をご参照ください。
password	認証に使用するパスワードです。	文字列	いいえ	なし
document-id.key-delimiter	ドキュメント ID のデリミターです。	文字列	いいえ	_	コネクタは、プライマリキーを使用してドキュメント ID を生成します。コネクタは、document-id.key-delimiter で指定されたデリミターを使用し、DDL で定義された順序でプライマリキーのすべてのフィールドを連結して、各行のドキュメント ID 文字列を作成します。説明ドキュメント ID は、スペースを含まない最大 512 バイトの文字列です。
failure-handler	失敗した Elasticsearch リクエストの障害処理ポリシーです。	文字列	いいえ	fail	有効なポリシー： `fail` (デフォルト)：リクエストが失敗した場合、ジョブは失敗します。 `ignore`：失敗を無視し、リクエストを破棄します。 `retry-rejected`：キューがいっぱいで失敗したリクエストを再追加します。カスタムクラス名：障害処理に ActionRequestFailureHandler のサブクラスを使用します。
sink.flush-on-checkpoint	チェックポイントでフラッシュするかどうかを指定します。	Boolean	いいえ	true	`true`：デフォルト値です。 `false`：無効にすると、チェックポイント中にコネクタは保留中のすべてのリクエストの確認応答を待機しません。つまり、コネクタは少なくとも 1 回の配信保証を提供しません。
sink.bulk-flush.backoff.strategy	一時的なリクエストエラーによってフラッシュ操作が失敗する場合は、sink.bulk-flush.backoff.strategy を設定してバックオフ戦略を指定します。	Enum	いいえ	DISABLED	`DISABLED` (デフォルト)：リトライは実行されません。最初のリクエストエラーでジョブが失敗します。 `CONSTANT`：リトライ間の待機時間が常に同じとなる、一定のバックオフ戦略です。 `EXPONENTIAL`：リトライ間の待機時間が指数関数的に増加する、エクスポネンシャルバックオフ戦略です。
sink.bulk-flush.backoff.max-retries	リトライの最大回数です。	Int	いいえ	なし	なし
sink.bulk-flush.backoff.delay	リトライ間の遅延時間です。	Duration	いいえ	なし	一定のバックオフ戦略の場合、この値は各リトライ間の遅延時間です。エクスポネンシャルバックオフ戦略の場合、この値は初期の基準遅延です。
sink.bulk-flush.max-actions	各バルクリクエストに対してバッファリングされるアクションの最大数です。	Int	いいえ	1000	値が 0 の場合、この機能は無効になります。
sink.bulk-flush.max-size	リクエストバッファーの最大メモリサイズです。	文字列	いいえ	2 MB	単位は MB です。デフォルト値は 2 MB です。値が 0 の場合、この機能は無効になります。
sink.bulk-flush.interval	フラッシュ間隔です。	Duration	いいえ	1 s	単位は秒です。デフォルト値は 1 s です。値が 0 s の場合、この機能は無効になります。
connection.path-prefix	各 REST 通信パスの先頭に付加される文字列です。	文字列	いいえ	なし	なし
retry-on-conflict	バージョンの競合が発生した場合に、更新操作をリトライする最大回数です。リトライ回数がこの値を超えると、例外が発生してジョブが失敗します。	Int	いいえ	0	説明このオプションは VVR 4.0.13 以降でのみサポートされています。このオプションが有効になるのは、プライマリキーが定義されている場合のみです。
routing-fields	ドキュメントを特定のシャードにルーティングするための、1 つ以上の Elasticsearch フィールド名を指定します。	文字列	いいえ	なし	複数のフィールド名はセミコロン (;) で区切ります。フィールドのデータが空の場合、フィールドは null に設定されます。説明このオプションは、`elasticsearch-7` および `elasticsearch-8` の場合に、VVR 8.0.6 以降でのみサポートされています。
sink.delete-strategy	シンクが撤回メッセージ (-D：DELETE または -U：UPDATE_BEFORE) をどのように処理するかを設定します。	Enum	いいえ	DELETE_ROW_ON_PK	有効な戦略： `DELETE_ROW_ON_PK` (デフォルト)：-U メッセージは無視しますが、-D メッセージを受信するとプライマリキーに対応する行 (ドキュメント) を削除します。 `IGNORE_DELETE`：-U と -D の両方のメッセージを無視します。Elasticsearch シンクでは撤回は発生しません。 NON_PK_FIELD_TO_NULL：-U メッセージを無視します。ただし、-D メッセージを受信した場合、この設定はプライマリキーに対応する行 (ドキュメント) を変更します。プライマリキーの値はそのままで、テーブルスキーマ内のプライマリキー以外の値はすべて NULL に設定されます。これは主に、複数のシンクが同じ Elasticsearch テーブルに同時に書き込む場合の部分更新に使用されます。 `CHANGELOG_STANDARD`：`DELETE_ROW_ON_PK` と似ていますが、-U メッセージを受信した場合もプライマリキーに対応する行 (ドキュメント) を削除します。説明このオプションは VVR 8.0.8 以降でのみサポートされています。
sink.ignore-null-when-update	データを更新する際に、受信したフィールド値が `null` の場合にフィールドを `null` に更新するか、変更せずに維持するかを指定します。	Boolean	いいえ	false	有効な値： `true`：フィールドは更新されません。この値は、Flink テーブルにプライマリキーが設定されており、Elasticsearch のデータフォーマットが JSON の場合にのみサポートされます。 `false`：フィールドは null に更新されます。説明このオプションは VVR 11.1 以降でのみサポートされています。
connection.request-timeout	接続マネージャーから接続を要求する際のタイムアウトです。	Duration	いいえ	なし	例： `'connection.request-timeout' = '1 min' -- 1 分 'connection.request-timeout' = '500ms' -- 500 ミリ秒` 説明このオプションは VVR 11.7 以降でのみサポートされています。
connect.timeout	接続を確立する際のタイムアウトです。	Duration	いいえ	なし	例： `'connect.timeout' = '1 min' -- 1 分 'connect.timeout' = '500ms' -- 500 ミリ秒` 説明このオプションは VVR 11.7 以降でのみサポートされています。
socket.timeout	データの待機に関するタイムアウトです。2 つの連続するデータパケット間の非アクティブ期間の最大値です。	Duration	いいえ	なし	例： `'socket.timeout' = '1 min' -- 1 分 'socket.timeout' = '500ms' -- 500 ミリ秒` 説明このオプションは VVR 11.7 以降でのみサポートされています。
connection.keep-alive	システムが接続を閉じるまでに、接続がアイドル状態を維持できる最大時間です。このオプションが設定されていない場合、サーバーの `Keep-Alive` 応答ヘッダーによって時間が決まります。サーバーが `Keep-Alive` ヘッダーを送信しない場合、接続は無期限に維持されます。	Duration	いいえ	なし	例： `'connection.keep-alive' = '1 min' -- 1 分 'connection.keep-alive' = '500ms' -- 500 ミリ秒` 説明このオプションは VVR 11.7 以降でのみサポートされています。
sink.bulk-flush.update.doc_as_upsert	更新リクエストでドキュメントを upsert ドキュメントとして扱うかどうかを指定します。	Boolean	いいえ	false	有効な値： `true`：更新リクエストの `doc_as_upsert` フィールドを true に設定します。 `false`：更新リクエストの `upsert` フィールドにドキュメントを設定します。 https://github.com/elastic/elasticsearch/issues/105804 によると、Elasticsearch のインジェストパイプラインは、バルク更新リクエストの部分更新をサポートしていません。インジェストパイプラインを使用する場合は、このオプションを true に設定してください。説明このオプションは VVR 11.5 以降でのみサポートされています。

ディメンションテーブル

パラメーター	説明	タイプ	必須	デフォルト	備考
connector	ディメンションテーブルのタイプ。	文字列	はい	なし	有効な値：`elasticsearch` または `elasticsearch-8`。説明 VVR 11.6 以降でのみ `elasticsearch-8` 値がサポートされます。
endPoint	Elasticsearch クラスターのサーバーアドレス。	文字列	はい	なし	レガシーオプション名。
hosts	Elasticsearch クラスターのサーバーアドレス。	文字列	はい	なし	`elasticsearch-8` で使用します。
indexName	インデックスの名前。	文字列	はい	なし	レガシーオプション名。
index	インデックスの名前。	文字列	はい	なし	`elasticsearch-8` で使用します。
accessId	認証用のユーザー名。	文字列	いいえ	なし	デフォルトでは、このパラメーターは空で、認証は行われません。accessId を指定する場合は、空でない accessKey を指定する必要があります。説明 `elasticsearch-8` では、`username` および `password` パラメーターが使用されます。重要ユーザー名とパスワードの漏洩を防ぐために、プロジェクト変数を使用することを推奨します。詳細については、「プロジェクト変数」をご参照ください。
username	認証用のユーザー名。	文字列	いいえ	なし
accessKey	認証用のパスワード。	文字列	いいえ	なし
password	認証用のパスワード。	文字列	いいえ	なし
typeNames	タイプの名前。	文字列	いいえ	_doc	Elasticsearch 7.0 以降では、このオプションを設定しないことを推奨します。
maxJoinRows	1 回のルックアップで結合する行の最大数。	整数	いいえ	1024	なし
cache	キャッシュ戦略。	文字列	いいえ	なし	有効な値： `ALL`：ディメンションテーブルのすべてのデータをキャッシュします。ジョブの開始前に、システムはディメンションテーブルのすべてのデータをキャッシュにロードします。以降のルックアップではキャッシュから直接データが応答されます。キーが見つからない場合、システムは存在しないと見なします。システムは TTL が期限切れになると、キャッシュ全体を再ロードします。 `LRU`：ディメンションテーブルのデータの一部をキャッシュします。ソーステーブルからレコードが到着すると、システムはまずキャッシュ内のデータを検索します。キャッシュミスの場合、物理ディメンションテーブルをクエリします。 `None`：キャッシュなし。
cacheSize	行数で指定されるキャッシュのサイズ。	Long	いいえ	100000	cacheSize パラメーターは、cache で LRU キャッシュポリシーが選択されている場合にのみ有効です。
cacheTTLMs	キャッシュの TTL (Time-To-Live)。	Long	いいえ	Long.MAX_VALUE	単位：ミリ秒。cacheTTLMs の動作は、cache の設定によって異なります。 cache が `LRU` の場合、cacheTTLMs はキャッシュエントリの TTL です。デフォルトでは、エントリは期限切れになりません。 cache が `ALL` の場合、cacheTTLMs はキャッシュを再ロードする間隔です。デフォルトでは、キャッシュは再ロードされません。
ignoreKeywordSuffix	STRING フィールドに自動的に追加される .keyword サフィックスを無視するかどうかを指定します。	ブール値	いいえ	false	互換性のため、Flink は Elasticsearch の `Text` 型を `STRING` に変換し、デフォルトでフィールド名に .keyword サフィックスを追加します。有効な値： `true`：サフィックスを無視します。サフィックスが Elasticsearch の `Text` 型フィールドとのマッチングを妨げる場合は、このオプションを `true` に設定してください。 `false`：サフィックスを無視しません。
cacheEmpty	物理ディメンションテーブルでのルックアップの空の結果をキャッシュするかどうかを指定します。	ブール値	いいえ	true	cacheEmpty パラメーターは、cache で LRU キャッシュポリシーが使用されている場合にのみ有効です。
queryMaxDocs	非プライマリキーディメンションテーブルの場合、Elasticsearch サーバーが各ルックアップクエリに対して返すドキュメントの最大数です。	整数	いいえ	10000	デフォルト値の 10,000 は、Elasticsearch サーバーがクエリごとに返すことができるドキュメントの最大数と一致します。この値はこの制限を超えることはできません。説明このオプションは VVR 8.0.8 以降でのみサポートされています。プライマリキーテーブルのデータは一意であるため、このオプションは非プライマリキーディメンションテーブルに対してのみ有効です。大きなデフォルト値はクエリの正確性を確保するのに役立ちますが、Elasticsearch クエリ中のメモリ使用量が増加します。メモリの問題が発生した場合は、この値を減らしてメモリ使用量を最適化できます。

型マッピング

Flink は Elasticsearch のデータを JSON として解析します。詳細については、「データ型マッピング」をご参照ください。

例

ソーステーブルの例

CREATE TEMPORARY TABLE elasticsearch_source (
  name STRING,
  location STRING,
  `value` FLOAT
) WITH (
  'connector' = 'elasticsearch-8',
  'hosts' = '<yourHosts>',
  'username' = '${secret_values.ak_id}',
  'password' = '${secret_values.ak_secret}',
  'index' = '<yourIndexName>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  name STRING,
  location STRING,
  `value` FLOAT
) WITH (
  'connector' ='blackhole'
);

INSERT INTO blackhole_sink
SELECT name, location, `value`
FROM elasticsearch_source;

ディメンションテーブルの例

CREATE TEMPORARY TABLE datagen_source (
  id STRING, 
  data STRING,
  proctime as PROCTIME()
) WITH (
  'connector' = 'datagen' 
);

CREATE TEMPORARY TABLE es_dim (
  id STRING,
  `value` FLOAT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'elasticsearch-8',
  'hosts' = '<yourHosts>',
  'username' = '${secret_values.ak_id}',
  'password' = '${secret_values.ak_secret}',
  'index' = '<yourIndexName>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  id STRING,
  data STRING,
  `value` FLOAT
) WITH (
  'connector' = 'blackhole' 
);

INSERT INTO blackhole_sink
SELECT e.*, w.*
FROM datagen_source AS e
JOIN es_dim FOR SYSTEM_TIME AS OF e.proctime AS w
ON e.id = w.id;

シンクテーブルの例 1

この例では、テキストベクトル化の後、テキストコンテンツを Elasticsearch に書き込みます。

説明

事前に Elasticsearch でインデックスマッピングを作成してください。embedding フィールドのデータ型を dense_vector に設定し、次元を指定してください。そうしないと、Elasticsearch はそれを通常の配列型として推論する可能性があります。

    CREATE TEMPORARY TABLE datagen_source (
      id STRING,
      content STRING,
      embedding ARRAY<FLOAT>
    ) WITH (
      'connector' = 'datagen'
    );

    CREATE TEMPORARY TABLE es_sink (
      id STRING,
      content STRING,
      embedding ARRAY<FLOAT>,
      PRIMARY KEY (id) NOT ENFORCED -- プライマリキーはオプションです。プライマリキーを定義した場合、その値がドキュメント ID になります。そうでない場合、ランダムなドキュメント ID が生成されます。
    ) WITH (
      'connector' = 'elasticsearch-8',
      'hosts' = '<yourHosts>',
      'index' = '<yourIndex>',
      'username' ='${secret_values.ak_id}',
      'password' ='${secret_values.ak_secret}'
    );

    INSERT INTO es_sink
    SELECT id, content, embedding
    FROM datagen_source;

シンクテーブルの例 2

このコネクタは、ROW、ARRAY、MAP などの複合型を Elasticsearch に書き込むことに対応しています。

CREATE TEMPORARY TABLE datagen_source(  
  id STRING,
    details ROW<  
        name STRING,  
        ages ARRAY<INT>,  
        attributes MAP<STRING, STRING>  
    >
) WITH (  
    'connector' = 'datagen'
);

CREATE TEMPORARY TABLE es_sink (
  id STRING,
    details ROW<  
        name STRING,  
        ages ARRAY<INT>,  
        attributes MAP<STRING, STRING>  
    >, 
  PRIMARY KEY (id) NOT ENFORCED  -- プライマリキーはオプションです。プライマリキーを定義した場合、その値がドキュメント ID になります。そうでない場合、ランダムなドキュメント ID が生成されます。
) WITH (
  'connector' = 'elasticsearch-6',
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>',
  'document-type' = '<yourElasticsearch.types>',
  'username' ='${secret_values.ak_id}',
  'password' ='${secret_values.ak_secret}'
);

INSERT INTO es_sink
SELECT id, details
FROM datagen_source;