データストリームの設定（Set up a data stream）

データストリームの設定

データストリームを設定するには、次の手順に従います：

• 1. インデックスライフサイクルポリシーを作成
• 2. コンポーネントテンプレートを作成
• 3. インデックステンプレートを作成
• 4. データストリームを作成
• 5. データストリームを保護

また、インデックスエイリアスをデータストリームに変換することもできます。

Fleet、Elastic Agent、またはLogstashを使用している場合は、このチュートリアルをスキップしてください。これらはすべて、データストリームを自動的に設定します。

FleetおよびElastic Agentについては、このデータストリームのドキュメントを確認してください。Logstashについては、elasticsearch outputプラグインのデータストリーム設定を確認してください。

インデックスライフサイクルポリシーの作成

オプションですが、ILMを使用してデータストリームのバックインデックスの管理を自動化することをお勧めします。ILMにはインデックスライフサイクルポリシーが必要です。

Kibanaでインデックスライフサイクルポリシーを作成するには、メインメニューを開き、**スタック管理

インデックスライフサイクルポリシーに移動します。ポリシーを作成**をクリックします。

また、ライフサイクルポリシー作成APIを使用することもできます。

Python

resp = client.ilm.put_lifecycle(
   name="my-lifecycle-policy",
   policy={
   "phases": {
   "hot": {
   "actions": {
   "rollover": {
   "max_primary_shard_size": "50gb"
   }
   }
   },
   "warm": {
   "min_age": "30d",
   "actions": {
   "shrink": {
   "number_of_shards": 1
   },
   "forcemerge": {
   "max_num_segments": 1
   }
   }
   },
   "cold": {
   "min_age": "60d",
   "actions": {
   "searchable_snapshot": {
   "snapshot_repository": "found-snapshots"
   }
   }
   },
   "frozen": {
   "min_age": "90d",
   "actions": {
   "searchable_snapshot": {
   "snapshot_repository": "found-snapshots"
   }
   }
   },
   "delete": {
   "min_age": "735d",
   "actions": {
   "delete": {}
   }
   }
   }
   },
)
print(resp)

Js

const response = await client.ilm.putLifecycle({
  name: "my-lifecycle-policy",
  policy: {
   phases: {
   hot: {
   actions: {
   rollover: {
   max_primary_shard_size: "50gb",
   },
   },
   },
   warm: {
   min_age: "30d",
   actions: {
   shrink: {
   number_of_shards: 1,
   },
   forcemerge: {
   max_num_segments: 1,
   },
   },
   },
   cold: {
   min_age: "60d",
   actions: {
   searchable_snapshot: {
   snapshot_repository: "found-snapshots",
   },
   },
   },
   frozen: {
   min_age: "90d",
   actions: {
   searchable_snapshot: {
   snapshot_repository: "found-snapshots",
   },
   },
   },
   delete: {
   min_age: "735d",
   actions: {
   delete: {},
   },
   },
   },
  },
});
console.log(response);

コンソール

PUT _ilm/policy/my-lifecycle-policy
{
  "policy": {
   "phases": {
   "hot": {
   "actions": {
   "rollover": {
   "max_primary_shard_size": "50gb"
   }
   }
   },
   "warm": {
   "min_age": "30d",
   "actions": {
   "shrink": {
   "number_of_shards": 1
   },
   "forcemerge": {
   "max_num_segments": 1
   }
   }
   },
   "cold": {
   "min_age": "60d",
   "actions": {
   "searchable_snapshot": {
   "snapshot_repository": "found-snapshots"
   }
   }
   },
   "frozen": {
   "min_age": "90d",
   "actions": {
   "searchable_snapshot": {
   "snapshot_repository": "found-snapshots"
   }
   }
   },
   "delete": {
   "min_age": "735d",
   "actions": {
   "delete": {}
   }
   }
   }
  }
}

コンポーネントテンプレートの作成

データストリームには、一致するインデックステンプレートが必要です。ほとんどの場合、1つ以上のコンポーネントテンプレートを使用してこのインデックステンプレートを構成します。通常、マッピングとインデックス設定には別々のコンポーネントテンプレートを使用します。これにより、複数のインデックステンプレートでコンポーネントテンプレートを再利用できます。

コンポーネントテンプレートを作成する際には、次を含めます：

• date](/read/elasticsearch-8-15/9dfa1da42eb162ff.md)またはdate_nanosマッピングを@timestampフィールドに対して指定します。マッピングを指定しない場合、Elasticsearchは@timestampをデフォルトオプションのdateフィールドとしてマッピングします。
• index.lifecycle.nameインデックス設定におけるライフサイクルポリシー。

フィールドをマッピングする際には、Elastic Common Schema (ECS)を使用してください。ECSフィールドは、デフォルトでいくつかのElastic Stack機能と統合されます。

フィールドのマッピング方法が不明な場合は、ランタイムフィールドを使用して、検索時に非構造化コンテンツからフィールドを抽出します。たとえば、ログメッセージをwildcardフィールドにインデックスし、後で検索中にこのフィールドからIPアドレスやその他のデータを抽出できます。

Kibanaでコンポーネントテンプレートを作成するには、メインメニューを開き、スタック
**管理

インデックス管理に移動します。インデックステンプレートビューで、コンポーネントテンプレートを作成**をクリックします。

また、コンポーネントテンプレート作成APIを使用することもできます。

Python

resp = client.cluster.put_component_template(
   name="my-mappings",
   template={
   "mappings": {
   "properties": {
   "@timestamp": {
   "type": "date",
   "format": "date_optional_time||epoch_millis"
   },
   "message": {
   "type": "wildcard"
   }
   }
   }
   },
   meta={
   "description": "Mappings for @timestamp and message fields",
   "my-custom-meta-field": "More arbitrary metadata"
   },
)
print(resp)
resp1 = client.cluster.put_component_template(
   name="my-settings",
   template={
   "settings": {
   "index.lifecycle.name": "my-lifecycle-policy"
   }
   },
   meta={
   "description": "Settings for ILM",
   "my-custom-meta-field": "More arbitrary metadata"
   },
)
print(resp1)

Ruby

response = client.cluster.put_component_template(
  name: 'my-mappings',
  body: {
   template: {
   mappings: {
   properties: {
   "@timestamp": {
   type: 'date',
   format: 'date_optional_time||epoch_millis'
   },
   message: {
   type: 'wildcard'
   }
   }
   }
   },
   _meta: {
   description: 'Mappings for @timestamp and message fields',
   "my-custom-meta-field": 'More arbitrary metadata'
   }
  }
)
puts response
response = client.cluster.put_component_template(
  name: 'my-settings',
  body: {
   template: {
   settings: {
   'index.lifecycle.name' => 'my-lifecycle-policy'
   }
   },
   _meta: {
   description: 'Settings for ILM',
   "my-custom-meta-field": 'More arbitrary metadata'
   }
  }
)
puts response

Js

const response = await client.cluster.putComponentTemplate({
  name: "my-mappings",
  template: {
   mappings: {
   properties: {
   "@timestamp": {
   type: "date",
   format: "date_optional_time||epoch_millis",
   },
   message: {
   type: "wildcard",
   },
   },
   },
  },
  _meta: {
   description: "Mappings for @timestamp and message fields",
   "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response);
const response1 = await client.cluster.putComponentTemplate({
  name: "my-settings",
  template: {
   settings: {
   "index.lifecycle.name": "my-lifecycle-policy",
   },
  },
  _meta: {
   description: "Settings for ILM",
   "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response1);

コンソール

# マッピング用のコンポーネントテンプレートを作成
PUT _component_template/my-mappings
{
  "template": {
   "mappings": {
   "properties": {
   "@timestamp": {
   "type": "date",
   "format": "date_optional_time||epoch_millis"
   },
   "message": {
   "type": "wildcard"
   }
   }
   }
  },
  "_meta": {
   "description": "@timestampおよびmessageフィールドのマッピング",
   "my-custom-meta-field": "より多くの任意のメタデータ"
  }
}
# インデックス設定用のコンポーネントテンプレートを作成
PUT _component_template/my-settings
{
  "template": {
   "settings": {
   "index.lifecycle.name": "my-lifecycle-policy"
   }
  },
  "_meta": {
   "description": "ILMの設定",
   "my-custom-meta-field": "より多くの任意のメタデータ"
  }
}

インデックステンプレートの作成

コンポーネントテンプレートを使用してインデックステンプレートを作成します。次を指定します：

• データストリームの名前に一致する1つ以上のインデックスパターン。私たちのデータストリーム命名スキームを使用することをお勧めします。
• テンプレートがデータストリーム対応であること。
• マッピングとインデックス設定を含む任意のコンポーネントテンプレート。
• 組み込みテンプレートとの衝突を避けるために、200よりも高い優先度。 インデックスパターンの衝突を避けるを参照してください。

Kibanaでインデックステンプレートを作成するには、メインメニューを開き、スタック
**管理

インデックス管理に移動します。インデックステンプレートビューで、テンプレートを作成**をクリックします。

また、インデックステンプレート作成APIを使用することもできます。data_streamオブジェクトを含めてデータストリームを有効にします。

Python

resp = client.indices.put_index_template(
   name="my-index-template",
   index_patterns=[
   "my-data-stream*"
   ],
   data_stream={},
   composed_of=[
   "my-mappings",
   "my-settings"
   ],
   priority=500,
   meta={
   "description": "Template for my time series data",
   "my-custom-meta-field": "More arbitrary metadata"
   },
)
print(resp)

Ruby

response = client.indices.put_index_template(
  name: 'my-index-template',
  body: {
   index_patterns: [
   'my-data-stream*'
   ],
   data_stream: {},
   composed_of: [
   'my-mappings',
   'my-settings'
   ],
   priority: 500,
   _meta: {
   description: 'Template for my time series data',
   "my-custom-meta-field": 'More arbitrary metadata'
   }
  }
)
puts response

Js

const response = await client.indices.putIndexTemplate({
  name: "my-index-template",
  index_patterns: ["my-data-stream*"],
  data_stream: {},
  composed_of: ["my-mappings", "my-settings"],
  priority: 500,
  _meta: {
   description: "Template for my time series data",
   "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response);

コンソール

PUT _index_template/my-index-template
{
  "index_patterns": ["my-data-stream*"],
  "data_stream": { },
  "composed_of": [ "my-mappings", "my-settings" ],
  "priority": 500,
  "_meta": {
   "description": "Template for my time series data",
   "my-custom-meta-field": "More arbitrary metadata"
  }
}

データストリームの作成

インデックスリクエストは、データストリームにドキュメントを追加します。これらのリクエストは、op_typeのcreateを使用する必要があります。ドキュメントには@timestampフィールドが含まれている必要があります。

データストリームを自動的に作成するには、ストリームの名前をターゲットにしたインデックスリクエストを送信します。この名前は、インデックステンプレートのインデックスパターンの1つと一致する必要があります。

Python

resp = client.bulk(
   index="my-data-stream",
   operations=[
   {
   "create": {}
   },
   {
   "@timestamp": "2099-05-06T16:21:15.000Z",
   "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
   },
   {
   "create": {}
   },
   {
   "@timestamp": "2099-05-06T16:25:42.000Z",
   "message": "192.0.2.255 - - [06/May/2099:16:25:42 +0000] \"GET /favicon.ico HTTP/1.0\" 200 3638"
   }
   ],
)
print(resp)
resp1 = client.index(
   index="my-data-stream",
   document={
   "@timestamp": "2099-05-06T16:21:15.000Z",
   "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
   },
)
print(resp1)

Ruby

response = client.bulk(
  index: 'my-data-stream',
  body: [
   {
   create: {}
   },
   {
   "@timestamp": '2099-05-06T16:21:15.000Z',
   message: '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736'
   },
   {
   create: {}
   },
   {
   "@timestamp": '2099-05-06T16:25:42.000Z',
   message: '192.0.2.255 - - [06/May/2099:16:25:42 +0000] "GET /favicon.ico HTTP/1.0" 200 3638'
   }
  ]
)
puts response
response = client.index(
  index: 'my-data-stream',
  body: {
   "@timestamp": '2099-05-06T16:21:15.000Z',
   message: '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736'
  }
)
puts response

Js

const response = await client.bulk({
  index: "my-data-stream",
  operations: [
   {
   create: {},
   },
   {
   "@timestamp": "2099-05-06T16:21:15.000Z",
   message:
   '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736',
   },
   {
   create: {},
   },
   {
   "@timestamp": "2099-05-06T16:25:42.000Z",
   message:
   '192.0.2.255 - - [06/May/2099:16:25:42 +0000] "GET /favicon.ico HTTP/1.0" 200 3638',
   },
  ],
});
console.log(response);
const response1 = await client.index({
  index: "my-data-stream",
  document: {
   "@timestamp": "2099-05-06T16:21:15.000Z",
   message:
   '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736',
  },
});
console.log(response1);

コンソール

PUT my-data-stream/_bulk
{ "create":{ } }
{ "@timestamp": "2099-05-06T16:21:15.000Z", "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736" }
{ "create":{ } }
{ "@timestamp": "2099-05-06T16:25:42.000Z", "message": "192.0.2.255 - - [06/May/2099:16:25:42 +0000] \"GET /favicon.ico HTTP/1.0\" 200 3638" }
POST my-data-stream/_doc
{
  "@timestamp": "2099-05-06T16:21:15.000Z",
  "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
}

また、データストリーム作成APIを使用して手動でストリームを作成することもできます。ストリームの名前は、テンプレートのインデックスパターンの1つと一致する必要があります。

Python

resp = client.indices.create_data_stream(
   name="my-data-stream",
)
print(resp)

Ruby

response = client.indices.create_data_stream(
  name: 'my-data-stream'
)
puts response

Js

const response = await client.indices.createDataStream({
  name: "my-data-stream",
});
console.log(response);

コンソール

PUT _data_stream/my-data-stream

データストリームの保護

インデックス権限を使用してデータストリームへのアクセスを制御します。データストリームに対する権限を付与すると、そのバックインデックスに対しても同じ権限が付与されます。

例については、データストリーム権限を参照してください。

インデックスエイリアスをデータストリームに変換

Elasticsearch 7.9以前では、書き込みインデックスを持つインデックスエイリアスを使用して時系列データを管理していました。データストリームはこの機能を置き換え、メンテナンスが少なく、データティアと自動的に統合されます。

書き込みインデックスを持つインデックスエイリアスを同じ名前のデータストリームに変換するには、データストリームへの移行APIを使用します。変換中に、エイリアスのインデックスはストリームの隠れたバックインデックスになります。エイリアスの書き込みインデックスはストリームの書き込みインデックスになります。ストリームには、データストリームが有効な一致するインデックステンプレートが必要です。

Python

resp = client.indices.migrate_to_data_stream(
   name="my-time-series-data",
)
print(resp)

Js

const response = await client.indices.migrateToDataStream({
  name: "my-time-series-data",
});
console.log(response);

コンソール

POST _data_stream/_migrate/my-time-series-data

データストリームに関する情報を取得

Kibanaでデータストリームに関する情報を取得するには、メインメニューを開き、**スタック管理

インデックス管理に移動します。データストリーム**ビューで、データストリームの名前をクリックします。

また、データストリーム取得APIを使用することもできます。

Python

resp = client.indices.get_data_stream(
   name="my-data-stream",
)
print(resp)

Ruby

response = client.indices.get_data_stream(
  name: 'my-data-stream'
)
puts response

Js

const response = await client.indices.getDataStream({
  name: "my-data-stream",
});
console.log(response);

コンソール

GET _data_stream/my-data-stream

データストリームの削除

Kibanaでデータストリームとそのバックインデックスを削除するには、メインメニューを開き、**スタック管理

インデックス管理に移動します。データストリーム**ビューで、ゴミ箱アイコンをクリックします。このアイコンは、データストリームに対するdelete_index セキュリティ権限を持っている場合にのみ表示されます。

また、データストリーム削除APIを使用することもできます。

Python

resp = client.indices.delete_data_stream(
   name="my-data-stream",
)
print(resp)

Ruby

response = client.indices.delete_data_stream(
  name: 'my-data-stream'
)
puts response

Js

const response = await client.indices.deleteDataStream({
  name: "my-data-stream",
});
console.log(response);

コンソール

DELETE _data_stream/my-data-stream