Elasticsearch 同期タスクの作成方法 - DataHub - Alibaba Cloud ドキュメントセンター

事前準備

Elasticsearch インデックスの作成
DataHub は、Elasticsearch インスタンスのインデックスへのデータ同期をサポートしています。この機能は、Elasticsearch のバージョン 5、6、および 7 と互換性があります。
現在、DataHub は TUPLE 型トピックのデータのみを Elasticsearch に同期できます。同期タスクを開始する前に、Elasticsearch にインデックスが作成されているか、インデックスの自動作成が有効になっていること を確認してください。そうでない場合、同期タスクは失敗します。
同期タスク用のアカウントの準備と権限付与
Elasticsearch 同期タスクを作成するには、Elasticsearch のエンドポイント、インデックスの詳細、およびアカウント認証情報を提供する必要があります。タスク作成の失敗を防ぐため、アカウント情報が有効であることを確認してください。

同期タスクの作成

[プロジェクト一覧] > [プロジェクト詳細] > [トピック詳細] ページに移動します。
右上隅の [+ Sync] をクリックします。
コネクタタイプとして Elasticsearch を選択します。

パラメータ

エンドポイント

Elasticsearch のサービスアドレスには、内部アドレス:内部ポート の形式で内部アドレスと内部ポートが必要です。

たとえば、内部アドレスが es-cn-xxx.elasticsearch.aliyuncs.com で、内部ポートが 9200 の場合、es-cn-xxx.elasticsearch.aliyuncs.com:9200 と入力します。

インデックス

宛先インデックスは、静的または動的の 2 つの方法で指定できます。

静的インデックス

すべてのデータは、単一の事前定義されたインデックスに書き込まれます。このインデックスを事前に作成するか、Elasticsearch でインデックスの自動作成を有効にする必要があります。

動的インデックス

データは、期間または特定のデータカラムの値に基づいて、異なるインデックスに書き込まれます。動的インデックスを使用するには、Elasticsearch でインデックスの自動作成を有効にする必要があります。インデックス名を生成するために、最大 1 つのカラムを選択できます。

サポートされる時間形式：

年	月	日	週
%Y	%m	%d	%U

例 1：毎日深夜に新しいインデックスを生成 インデックスを test_${%Y-%m-%d} として設定します。現在の日付が 2021 年 3 月 31 日の場合、最終的なインデックスは test_2021-03-31 になります。
例 2：データカラムから新しいインデックスを生成 データに col1 という名前のカラムが含まれ、インデックスが test_${col1} として設定されているとします。2 つのデータレコードの col1 カラムの値が AAA と BBB の場合、レコードはそれぞれ test_AAA インデックスと test_BBB インデックスに書き込まれます。

Elasticsearch のインデックス数が増加すると、書き込みパフォーマンスが低下する可能性があります。インデックス数が多すぎると、DataHub で書き込みタイムアウトが発生する場合があります。動的インデックスを使用する場合は、インデックスが過剰に作成されないように命名パターンを設計してください。

ユーザー/パスワード

Elasticsearch にアクセスするためのユーザー名とパスワードです。

タイプカラム

DataHub は、Elasticsearch のバージョンに応じてドキュメントタイプを異なる方法で生成します。Elasticsearch 5 では、1 つのインデックス内に複数のタイプを作成できます。Elasticsearch 6 では、インデックスごとに 1 つのタイプしか作成できません。この設定を空にすることはできません。

Elasticsearch 5 の場合、DataHub は選択されたタイプカラムの値をドキュメントタイプとして使用します。複数のカラムを選択した場合、それらの値が "|" 区切り文字で連結されてタイプが作成されます。タイプカラムに選択されたフィールドに null 値を含めることはできません。
Elasticsearch 6 の場合、DataHub は選択されたカラムの名前をドキュメントタイプとして使用します。複数のカラムを選択した場合、それらの名前が "|" 区切り文字で連結されます。Elasticsearch 6 では、任意の文字列をタイプ名として使用できます。
Elasticsearch 6 の同期タスクを作成する際、コンソールでカスタムタイプ名を指定することはできません。カスタムタイプ名を使用するには、SDK を使用してタスクを作成する必要があります。
Elasticsearch 7 の場合、すべてのドキュメントはデフォルトのタイプを使用するため、タイプカラムを選択する必要はありません。

例：

DataHub スキーマ：f1 string, f2 string, f3 string, f4 string
データレコード：["test1","test2","test3",null]

タイプカラム	Elasticsearch 5 のタイプ	Elasticsearch 6 のタイプ
f1	test1	f1
f1、f3	test1\|test3	f1\|f3
ff	作成に失敗します	ff
f1、ff	作成に失敗します	f1\|ff
f4	作成は成功しますが、同期は失敗します (ダーティデータ)	作成と同期に成功します

ID カラム

DataHub のデータレコードから Elasticsearch のドキュメント ID を生成できます。カラムを選択しない場合、Elasticsearch は各ドキュメントに一意の ID を生成します。1 つ以上のカラムを選択した場合、それらの値が "|" 区切り文字で連結されてドキュメント ID が作成されます。ID カラムに選択されたフィールドに null 値を含めることはできません。

例：

DataHub スキーマ：f1 string, f2 string, f3 string, f4 string
データレコード：["test1","test2","test3",null]

ID カラム	ドキュメント ID
	Elasticsearch が一意の ID を自動生成します
f1	test1
f1、f3	test1\|test3
ff	作成に失敗します
f4	作成は成功しますが、同期は失敗します (ダーティデータ)

ルーティングカラム

DataHub のデータレコードから Elasticsearch のルーティング値を生成できます。カラムを選択しない場合、Elasticsearch のルーティング機能は使用されません。1 つ以上のカラムを選択した場合、それらの値がルーティング値として使用されます。複数のカラムを選択した場合、それらの値が "|" 区切り文字で連結されます。ルーティングカラムに選択されたフィールドに null 値を含めることはできません。

例については、ID カラム をご参照ください。

インポートフィールド

DataHub トピックから Elasticsearch に同期するフィールドを選択します。DataHub は、選択されていないフィールドを同期しません。Elasticsearch 5 の場合、最終的なドキュメントデータには、ID カラムおよびタイプカラムに使用されるフィールドは含まれません。データ変換の詳細については、以降のセクションの例をご参照ください。

ネットワークタイプ

Elasticsearch インスタンスのデプロイに基づいてネットワークタイプを選択します。パブリッククラウド上の Alibaba Cloud Elasticsearch インスタンスは VPC 内で実行されます。したがって、これらの同期タスクのネットワークタイプとして VPC を選択します。VPC ネットワークタイプを使用する場合、VPC ID とインスタンス ID を指定する必要があります。

インスタンス ID を入力する際は、サフィックス -worker を追加する必要があります。たとえば、インスタンス ID が es-cn-xxx の場合、 es-cn-xxx-worker と入力する必要があります。

データ書き込みの例

以下の例では、Elasticsearch 同期タスクが正常に作成されていることを前提としています。タスクの作成に失敗した場合は、設定を確認して修正してください。

DataHub スキーマ：

フィールド名	フィールドタイプ
f1	BIGINT
f2	STRING
f3	BOOLEAN
f4	DOUBLE
f5	TIMESTAMP
f6	DECIMAL

例 1

タイプ列 = f1 (Elasticsearch 7 には適用されません)
ID 列 = f2

インポートされたフィールド = f1, f2, f3, f4, f5, f6

データレコード = v1, v2, v3, v4, v5, v6
ES バージョン
タイプ
Id
データ
ES5
v1
v2
{f3:v3,f4:v4,f5:v5,f6:v6}
ES6
f1
v2
{f1:v1,f3:v3,f4:v4,f5:v5,f6:v6}
ES7
-
v2
{f1:v1,f3:v3,f4:v4,f5:v5,f6:v6}
データレコード = null, v2, v3, v4, v5, v6
ES バージョン
タイプ
Id
データ
ES5
-
-
タイプ列に NULL 値があると、ダーティデータが発生します。
ES6
f1
v2
{f3:v3,f4:v4,f5:v5,f6:v6}
ES7
-
v2
{f3:v3,f4:v4,f5:v5,f6:v6}

データレコード = v1, null, v3, v4, v5, v6

ES バージョン	タイプ	Id	データ
ES5	-	-	ID 列に NULL 値があると、ダーティデータが発生します。
ES6	-	-	ID 列に NULL 値があると、ダーティデータが発生します。
ES7	-	-	ID 列に NULL 値があると、ダーティデータが発生します。

例 2

タイプ列 = f1, f2 (Elasticsearch 7 には適用されません)
ID 列 = f3, f4

インポートされたフィールド = f1, f2, f3, f4, f5, f6

データレコード = v1, v2, v3, v4, v5, v6
ES バージョン
タイプ
Id
データ
ES5
v1|v2
v3|v4
{f5:v5,f6:v6}
ES6
f1|f2
v3|v4
{f1:v1,f2:v2,f5:v5,f6:v6}
ES7
-
v3|v4
{f1:v1,f2:v2,f5:v5,f6:v6}
データレコード = v1, null, v3, v4, v5, v6
ES バージョン
タイプ
Id
データ
ES5
-
-
タイプ列に NULL 値があると、ダーティデータが発生します。
ES6
f1|f2
v3|v4
{f1:v1,f5:v5,f6:v6}
ES7
-
v3|v4
{f1:v1,f5:v5,f6:v6}

データレコード = v1, v2, null, v4, v5, v6

ES バージョン	タイプ	Id	データ
ES5	-	-	ID 列に NULL 値があると、ダーティデータが発生します。
ES6	-	-	ID 列に NULL 値があると、ダーティデータが発生します。
ES7	-	-	ID 列に NULL 値があると、ダーティデータが発生します。

例 3

タイプ列 = f1 (Elasticsearch 7 には適用されません)
ID 列 = f2
ルーター列 = f3

インポートされたフィールド = f1, f2, f3, f4, f5, f6

データレコード = v1, v2, v3, v4, v5, v6
ES バージョン
タイプ
Id
ルーター
データ
ES5
v1
v2
v3
{f4:v4,f5:v5,f6:v6}
ES6
f1
v2
v3
{f1:v1,f4:v4,f5:v5,f6:v6}
ES7
-
v2
v3
{f1:v1,f4:v4,f5:v5,f6:v6}
データレコード = null, v2, v3, v4, v5, v6
ES バージョン
タイプ
Id
データ
ES5
-
-
タイプ列に NULL 値があると、ダーティデータが発生します。
ES6
f1
v2
{f4:v4,f5:v5,f6:v6}
ES7
-
v2
{f4:v4,f5:v5,f6:v6}

データレコード = v1, null, v3, v4, v5, v6

ES バージョン	タイプ	Id	データ
ES5	-	-	ID 列に NULL 値があると、ダーティデータが発生します。
ES6	-	-	ID 列に NULL 値があると、ダーティデータが発生します。
ES7	-	-	ID 列に NULL 値があると、ダーティデータが発生します。

データレコード = v1, v2, null, v4, v5, v6

ES バージョン	タイプ	Id	データ
ES5	-	-	ルーター列に NULL 値があると、ダーティデータが発生します。
ES6	-	-	ルーター列に NULL 値があると、ダーティデータが発生します。
ES7	-	-	ルーター列に NULL 値があると、ダーティデータが発生します。

同期タスクの管理

コネクタの詳細ページでは、タスクの実行ステータス、チェックポイント、その他の情報を表示できます。また、タスクの再起動と停止も行えます。チェックポイントをリセットする前に、タスクを停止する必要があります。

エンドツーエンドの例

この例では、Alibaba Cloud Elasticsearch 6.7 を使用して、DataHub から Elasticsearch にデータを同期する一連のプロセスを説明します。Elasticsearch 関連の操作は、Kibana Dev Tools を使用して実行します。他の方法については、公式 Elasticsearch ドキュメントをご参照ください。

Elasticsearch インデックスの作成
デフォルトでは、Elasticsearch は自動的にインデックスを作成するため、通常このステップはスキップできます。インデックスの自動作成が無効になっている場合、インデックスを手動で作成する必要があります。特定のコマンドについては、公式 Elasticsearch ドキュメントをご参照ください。
DataHub トピックの作成
TUPLE 型のトピックのみが Elasticsearch への同期に対応しています。DataHub トピックの作成方法については、「トピックの操作」をご参照ください。
Elasticsearch 同期タスクの作成
この例では、同期タスクの作成時に、タイプカラムとして f1 と f2 を、ID カラムとして f3 と f4 を使用し、すべてのフィールドをインポート対象として選択します。
DataHub トピックへのデータ書き込み
DataHub SDK またはプラグインを使用してデータを書き込みます。レコードを書き込んだ後、コンソールでデータをサンプリングして、書き込んだ内容を確認できます。
データ同期の検証
まず、Elasticsearch 同期タスクのチェックポイントを確認します。チェックポイントと同期時刻の変更は、データが同期されたことを意味します。同期時刻はデータが DataHub に書き込まれた時刻を反映し、チェックポイントの値が 1 の場合は、最初のデータレコード (インデックス 0) が書き込まれたことを意味します。
次に、Elasticsearch のデータを確認します。Kibana を使用して、データが正常に同期されたことを確認できます。

ES バージョン	タイプ	Id	ルーター	データ
ES5	v1	v2	v3	`{f4:v4,f5:v5,f6:v6}`
ES6	f1	v2	v3	`{f1:v1,f4:v4,f5:v5,f6:v6}`
ES7	-	v2	v3	`{f1:v1,f4:v4,f5:v5,f6:v6}`

ES バージョン	タイプ	Id	データ
ES5	v1\|v2	v3\|v4	`{f5:v5,f6:v6}`
ES6	f1\|f2	v3\|v4	{f1:v1,f2:v2,f5:v5,f6:v6}
ES7	-	v3\|v4	{f1:v1,f2:v2,f5:v5,f6:v6}