수집 파이프라인
원본 보기수집 파이프라인
Elasticsearch 수집 파이프라인을 사용하면 데이터를 인덱싱하기 전에 일반적인 변환을 수행할 수 있습니다. 예를 들어 파이프라인을 사용하여 필드를 제거하고, 텍스트에서 값을 추출하며, 데이터를 보강할 수 있습니다.
파이프라인은 프로세서라고 하는 구성 가능한 일련의 작업으로 이루어집니다. 각 프로세서는 순차적으로 실행되며 수신 문서를 구체적으로 변경합니다. 프로세서 실행이 끝나면 Elasticsearch가 변환된 문서를 데이터 스트림 또는 인덱스에 추가합니다.
Kibana의 수집 파이프라인 기능 또는 수집 API를 사용하여 수집 파이프라인을 생성하고 관리할 수 있습니다. Elasticsearch는 파이프라인을 클러스터 상태에 저장합니다.
ingest노드 역할이 있는 노드가 파이프라인 처리를 담당합니다. 수집 파이프라인을 사용하려면 클러스터에ingest역할인 노드가 하나 이상 있어야 합니다. 수집 부하가 크다면 전용 수집 노드를 생성하는 것이 좋습니다.- Elasticsearch 보안 기능이 활성화된 경우 수집 파이프라인을 관리하려면
manage_pipeline클러스터 권한 이 있어야 합니다. Kibana의 수집 파이프라인 기능을 사용하려면cluster:monitor/nodes/info클러스터 권한도 필요합니다. enrich프로세서가 포함된 파이프라인에는 추가 설정이 필요합니다. 자세한 내용은 데이터 보강을 참조하세요.
Kibana의 수집 파이프라인 관리 페이지로 이동하려면 탐색 메뉴 또는 전역 검색 필드를 사용합니다.
목록 보기에서는 다음 작업을 수행할 수 있습니다.
- 파이프라인 목록을 보고 세부 정보 살펴보기
- 기존 파이프라인 편집 또는 복제
- 파이프라인 삭제
파이프라인을 생성하려면 파이프라인 생성 > 새 파이프라인을 클릭합니다. 예제 튜토리얼은 예: 로그 구문 분석을 참조하세요.
CSV에서 새 파이프라인 생성 옵션을 사용하면 CSV로 사용자 지정 데이터를 Elastic Common Schema(ECS)에 매핑하는 수집 파이프라인을 생성할 수 있습니다. 사용자 지정 데이터를 ECS에 매핑하면 데이터를 더 쉽게 검색하고 다른 데이터 세트의 시각화를 재사용할 수 있습니다. 시작하려면 사용자 지정 데이터를 ECS에 매핑을 참조하세요.
또한 수집 API 를 사용하여 파이프라인을 생성하고 관리할 수 있습니다. 다음 파이프라인 생성 API 요청은 두 개의 set 프로세서와 그 뒤에 오는 lowercase 프로세서가 포함된 파이프라인을 생성합니다. 프로세서는 지정된 순서대로 실행됩니다.
PUT _ingest/pipeline/my-pipeline
{
"description": "My optional pipeline description",
"processors": [
{
"set": {
"description": "My optional processor description",
"field": "my-long-field",
"value": 10
}
},
{
"set": {
"description": "Set 'my-boolean-field' to true",
"field": "my-boolean-field",
"value": true
}
},
{
"lowercase": {
"field": "my-keyword-field"
}
}
]
}
파이프라인을 생성하거나 업데이트할 때 선택적으로 version 정수를 지정할 수 있습니다. 이 버전 번호를 if_version 매개변수와 함께 사용하여 조건부로 파이프라인을 업데이트할 수 있습니다. if_version 매개변수가 지정된 경우 업데이트가 성공하면 파이프라인 버전이 증가합니다.
PUT _ingest/pipeline/my-pipeline-id
{
"version": 1,
"processors": [ ... ]
}
API를 사용하여 version 번호 설정을 해제하려면 version 매개변수를 지정하지 않고 파이프라인을 교체하거나 업데이트합니다.
프로덕션에서 파이프라인을 사용하기 전에 샘플 문서로 테스트하는 것이 좋습니다. Kibana에서 파이프라인을 생성하거나 편집할 때 문서 추가를 클릭합니다. 문서 탭에서 샘플 문서를 제공하고 파이프라인 실행을 클릭합니다.
또한 파이프라인 시뮬레이션 API를 사용하여 파이프라인을 테스트할 수 있습니다. 요청 경로에 구성된 파이프라인을 지정할 수 있습니다. 예를 들어 다음 요청은 my-pipeline.
POST _ingest/pipeline/my-pipeline/_simulate
{
"docs": [
{
"_source": {
"my-keyword-field": "FOO"
}
},
{
"_source": {
"my-keyword-field": "BAR"
}
}
]
}
또는 요청 본문에 파이프라인과 해당 프로세서를 지정할 수 있습니다.
POST _ingest/pipeline/_simulate
{
"pipeline": {
"processors": [
{
"lowercase": {
"field": "my-keyword-field"
}
}
]
},
"docs": [
{
"_source": {
"my-keyword-field": "FOO"
}
},
{
"_source": {
"my-keyword-field": "BAR"
}
}
]
}
API는 변환된 문서를 반환합니다.
{
"docs": [
{
"doc": {
"_index": "_index",
"_id": "_id",
"_version": "-3",
"_source": {
"my-keyword-field": "foo"
},
"_ingest": {
"timestamp": "2099-03-07T11:04:03.000Z"
}
}
},
{
"doc": {
"_index": "_index",
"_id": "_id",
"_version": "-3",
"_source": {
"my-keyword-field": "bar"
},
"_ingest": {
"timestamp": "2099-03-07T11:04:04.000Z"
}
}
}
]
}
다음 pipeline 쿼리 매개변수를 사용하여 개별 또는 대량 인덱싱 요청의 문서에 파이프라인을 적용합니다.
POST my-data-stream/_doc?pipeline=my-pipeline
{
"@timestamp": "2099-03-07T11:04:05.000Z",
"my-keyword-field": "foo"
}
PUT my-data-stream/_bulk?pipeline=my-pipeline
{ "create":{ } }
{ "@timestamp": "2099-03-07T11:04:06.000Z", "my-keyword-field": "foo" }
{ "create":{ } }
{ "@timestamp": "2099-03-07T11:04:07.000Z", "my-keyword-field": "bar" }
또한 pipeline 매개변수를 쿼리로 업데이트 또는 재인덱싱 API와 함께 사용할 수도 있습니다.
POST my-data-stream/_update_by_query?pipeline=my-pipeline
POST _reindex
{
"source": {
"index": "my-data-stream"
},
"dest": {
"index": "my-new-data-stream",
"op_type": "create",
"pipeline": "my-pipeline"
}
}
다음 index.default_pipeline 인덱스 설정을 사용하여 기본 파이프라인을 설정합니다. Elasticsearch는 pipeline 매개변수가 지정되지 않은 인덱싱 요청에 이 파이프라인을 적용합니다.
다음 index.final_pipeline 인덱스 설정을 사용하여 최종 파이프라인을 설정합니다. Elasticsearch는 요청 파이프라인이나 기본 파이프라인 뒤에 이 파이프라인을 적용하며, 둘 다 지정되지 않은 경우에도 적용합니다.
Elastic Beat에 수집 파이프라인을 추가하려면 pipeline 아래에 output.elasticsearch 의 <BEAT_NAME>.yml를 지정합니다. 예를 들어 Filebeat의 경우 pipeline 의 filebeat.yml.
output.elasticsearch:
hosts: ["localhost:9200"]
pipeline: my-pipeline
Elastic Agent 통합에는 인덱싱 전에 데이터를 전처리하고 보강하는 기본 수집 파이프라인이 포함되어 있습니다. Fleet 는 인덱스 템플릿 을 사용해 이러한 파이프라인을 적용합니다. 여기에는 파이프라인 인덱스 설정이 포함됩니다. Elasticsearch는 스트림 명명 체계.
각 기본 통합 파이프라인은 존재하지 않으며 버전이 없는 *@custom 수집 파이프라인을 호출합니다. 변경하지 않으면 이 파이프라인 호출은 데이터에 영향을 주지 않습니다. 하지만 이 호출을 수정하여 업그레이드 후에도 유지되는 통합용 사용자 지정 파이프라인을 생성할 수 있습니다. 자세한 내용은 사용자 지정 수집 파이프라인으로 데이터 변환 을 참조하세요.
Fleet은 사용자 지정 로그 통합에 기본 수집 파이프라인을 제공하지 않지만 인덱스 템플릿 또는 사용자 지정 구성.
생성 하고 테스트 수집 파이프라인을 테스트합니다. 파이프라인 이름을
logs-<dataset-name>-default로 지정하면 통합의 파이프라인을 더 쉽게 추적할 수 있습니다.예를 들어 다음 요청은
my-app데이터 세트용 파이프라인을 생성합니다. 파이프라인 이름은logs-my_app-default.PUT _ingest/pipeline/logs-my_app-default{ "description": "Pipeline for `my_app` dataset", "processors": [ ... ] }다음 인덱스 템플릿 을 생성하고 파이프라인을
index.default_pipeline또는index.final_pipeline인덱스 설정에 포함합니다. 템플릿에서 데이터 스트림을 활성화해야 합니다. 템플릿의 인덱스 패턴은logs-<dataset-name>-*.이 템플릿은 Kibana의 인덱스 관리 기능 또는 인덱스 템플릿 생성 API.
예를 들어 다음 요청은
logs-my_app-*와 일치하는 템플릿을 생성합니다. 이 템플릿은index.default_pipeline인덱스 설정이 포함된 컴포넌트 템플릿을 사용합니다.# Creates a component template for index settingsPUT _component_template/logs-my_app-settings{ "template": { "settings": { "index.default_pipeline": "logs-my_app-default", "index.lifecycle.name": "logs" } } } # Creates an index template matching `logs-my_app-*`PUT _index_template/logs-my_app-template{ "index_patterns": ["logs-my_app-*"], "data_stream": { }, "priority": 500, "composed_of": ["logs-my_app-settings", "logs-my_app-mappings"] }Fleet에서 사용자 지정 로그 통합을 추가하거나 편집할 때 통합 구성 > 사용자 지정 로그 파일 > 고급 옵션.
다음 데이터 세트 이름에 데이터 세트 이름을 지정합니다. Fleet은 통합의 새 데이터를 결과
logs-<dataset-name>-default데이터 스트림에 추가합니다.예를 들어 데이터 세트 이름이
my_app이면 Fleet은 새 데이터를logs-my_app-default데이터 스트림에 추가합니다.
다음 롤오버 API 를 사용하여 데이터 스트림을 롤오버합니다. 그러면 Elasticsearch가 인덱스 템플릿과 해당 파이프라인 설정을 통합의 모든 새 데이터에 적용합니다.
POST logs-my_app-default/_rollover/
생성 하고 테스트 수집 파이프라인을 테스트합니다. 파이프라인 이름을
logs-<dataset-name>-default로 지정하면 통합의 파이프라인을 더 쉽게 추적할 수 있습니다.예를 들어 다음 요청은
my-app데이터 세트용 파이프라인을 생성합니다. 파이프라인 이름은logs-my_app-default.PUT _ingest/pipeline/logs-my_app-default{ "description": "Pipeline for `my_app` dataset", "processors": [ ... ] }Fleet에서 사용자 지정 로그 통합을 추가하거나 편집할 때 통합 구성 > 사용자 지정 로그 파일 > 고급 옵션.
다음 데이터 세트 이름에 데이터 세트 이름을 지정합니다. Fleet은 통합의 새 데이터를 결과
logs-<dataset-name>-default데이터 스트림에 추가합니다.예를 들어 데이터 세트 이름이
my_app이면 Fleet은 새 데이터를logs-my_app-default데이터 스트림에 추가합니다.다음 사용자 지정 구성에서 파이프라인을
pipeline정책 설정에 지정합니다.
독립 실행형 Elastic Agent
Elastic Agent를 독립 실행형으로 실행하는 경우 인덱스 템플릿 을 포함하는 인덱스 템플릿을 사용하여 파이프라인을 적용할 수 있습니다. index.default_pipeline 또는 index.final_pipeline 인덱스 설정. 또는 pipeline 정책 설정을 elastic-agent.yml 구성에 지정할 수 있습니다. 자세한 내용은 독립 실행형 Elastic Agent 설치.
예를 들어 웹 크롤러^ 또는 커넥터를 사용하여 검색 사용 사례용 Elasticsearch 인덱스를 생성하면 해당 인덱스에 특정 수집 파이프라인이 자동으로 설정됩니다. 이 프로세서는 검색에 맞게 콘텐츠를 최적화하는 데 도움이 됩니다. 자세한 내용은 Search의 수집 파이프라인 을 참조하세요.
프로세서는 수신 문서의 소스 필드에 대한 읽기 및 쓰기 권한이 있습니다. 프로세서에서 필드 키에 액세스하려면 필드 이름을 사용합니다. 다음 set 프로세서는 다음에 액세스합니다. my-long-field.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"field": "my-long-field",
"value": 10
}
}
]
}
앞에 _source 접두사를 추가할 수도 있습니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"field": "_source.my-long-field",
"value": 10
}
}
]
}
객체 필드에 액세스하려면 점 표기법을 사용합니다.
문서에 평면화된 객체가 포함된 경우 dot_expander 프로세서를 사용하여 확장합니다. 문서 구조를 유지하려면 파이프라인 정의에서 flexible 액세스 패턴을 사용합니다. 그렇지 않으면 수집 프로세서가 점으로 구분된 필드 이름에 액세스할 수 없습니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"dot_expander": {
"description": "Expand 'my-object-field.my-property'",
"field": "my-object-field.my-property"
}
},
{
"set": {
"description": "Set 'my-object-field.my-property' to 10",
"field": "my-object-field.my-property",
"value": 10
}
}
]
}
여러 프로세서 매개변수가 Mustache 템플릿 스니펫을 지원합니다. 템플릿 스니펫에서 필드 값에 액세스하려면 필드 이름을 삼중 중괄호로 묶습니다.{{{field-name}}}템플릿 스니펫을 사용하여 필드 이름을 동적으로 설정할 수 있습니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "Set dynamic '<service>' field to 'code' value",
"field": "{{{service}}}",
"value": "{{{code}}}"
}
}
]
}
기본 수집 파이프라인 액세스 패턴은 문서에서 점으로 구분된 필드 이름을 인식하지 못합니다. 수집 문서에서 평면화되고 점으로 구분된 필드 이름을 검색하려면 이러한 제한이 없는 다른 필드 검색 알고리즘이 필요합니다. 일부 파이프라인은 로직에서 이러한 제한에 의존합니다. 기존 동작을 계속 지원하면서 점으로 구분된 필드 이름도 지원할 수 있도록, 이제 수집 파이프라인에서 파이프라인의 모든 프로세서가 사용할 액세스 패턴을 구성할 수 있습니다.
이 field_access_pattern 속성은 현재 파이프라인의 모든 프로세서가 수집 문서 필드를 읽고 쓰는 방식을 정의합니다. 허용되는 값은 다음 두 가지입니다. classic 기본값인 flexible.
PUT _ingest/pipeline/my-pipeline
{
"field_access_pattern": "classic",
"processors": [
{
"set": {
"description": "Set some searchable tags in our document's flattened field",
"field": "event.tags.ingest.processed_by",
"value": "my-pipeline"
}
}
]
}
- 이 파이프라인의 모든 프로세서는
classic액세스 패턴을 사용합니다. - 프로세서가 수집 문서의 값을 읽고 쓸 때 사용하는 필드 경로 확인 로직은 액세스 패턴을 기반으로 합니다.
이 classic 액세스 패턴은 수집 노드가 처음 출시된 이후 사용된 기본 액세스 패턴입니다. 프로세서에 제공된 필드 경로(예: event.tags.ingest.processed_by)는 점 문자(.)를 기준으로 분할됩니다. 그런 다음 프로세서는 결과 필드 이름을 사용하여 값을 찾을 때까지 문서를 탐색합니다. 문서에 값을 쓸 때 상위 필드가 소스에 없으면 누락된 필드에 대한 중첩 객체를 생성합니다.
POST /_ingest/pipeline/_simulate
{
"pipeline" : {
"description": "example pipeline",
"field_access_pattern": "classic",
"processors": [
{
"set" : {
"description" : "Copy the foo.bar field into the a.b.c.d field if it exists",
"copy_from" : "foo.bar",
"field" : "a.b.c.d",
"ignore_empty_value": true
}
}
]
},
"docs": [
{
"_index": "index",
"_id": "id",
"_source": {
"foo": {
"bar": "baz"
}
}
},
{
"_index": "index",
"_id": "id",
"_source": {
"foo.bar": "baz"
}
}
]
}
- 파이프라인에서
classic액세스 패턴을 사용하도록 명시적으로 선언합니다. 이 값이 기본값입니다. - 다음 필드에서 값을 읽습니다.
foo.bar. - 값을 다음 필드에 씁니다.
a.b.c.d. - 이 문서 구조는 중첩 JSON 객체를 사용합니다.
- 이 문서 구조는 점으로 구분된 필드 이름을 사용합니다.
{
"docs": [
{
"doc": {
"_id": "id",
"_index": "index",
"_version": "-3",
"_source": {
"foo": {
"bar": "baz"
},
"a": {
"b": {
"c": {
"d": "baz"
}
}
}
},
"_ingest": {
"timestamp": "2017-05-04T22:30:03.187Z"
}
}
},
{
"doc": {
"_id": "id",
"_index": "index",
"_version": "-3",
"_source": {
"foo.bar": "baz"
},
"_ingest": {
"timestamp": "2017-05-04T22:30:03.188Z"
}
}
}
]
}
- 첫 번째 문서의
foo.bar필드는 중첩 JSON을 사용하므로 찾을 수 있습니다. 프로세서는 먼저foo필드를 찾은 다음bar필드를 찾습니다. - 다음 필드의 값은
foo.bar필드의 중첩 JSON 구조에 기록됩니다.a.b.c.d프로세서는 경로의 각 필드에 대한 객체를 생성합니다. - 두 번째 문서는 다음에 점으로 구분된 필드 이름을 사용합니다.
foo.bar다음classic액세스 패턴은 점으로 구분된 필드 이름을 인식하지 못하므로 아무것도 복사되지 않습니다.
수집하는 문서에 점으로 구분된 필드 이름이 포함되어 있고 이를 classic 액세스 패턴으로 읽으려면 dot_expander 프로세서를 사용해야 합니다. 하지만 이 방법이 항상 적합한 것은 아닙니다. 다음 문서를 살펴보세요.
{
"event": {
"tags": {
"http.host": "localhost:9200",
"http.host.name": "localhost",
"http.host.port": 9200
}
}
}
다음 event.tags 필드를 dot_expander 프로세서로 처리하면 필드 값이 충돌합니다. http.host 필드는 텍스트 값과 객체 값일 수 없습니다.
이 flexible 액세스 패턴을 사용하면 수집 파이프라인이 dot_expander 프로세서를 사용하지 않고 중첩 필드 이름과 점으로 구분된 필드 이름 모두에 액세스할 수 있습니다. 또한 존재하지 않는 필드에 값을 쓸 때 누락된 상위 필드는 새 키의 시작 부분에 연결됩니다. 문서에 점으로 구분된 필드 이름이 있거나 누락된 필드를 점으로 구분된 이름으로 문서에 쓰려면 flexible 액세스 패턴을 사용합니다.
POST /_ingest/pipeline/_simulate
{
"pipeline" : {
"description": "example pipeline",
"field_access_pattern": "flexible",
"processors": [
{
"set" : {
"description" : "Copy the foo.bar field into the a.b.c.d field if it exists",
"copy_from" : "foo.bar",
"field" : "a.b.c.d",
"ignore_empty_value": true
}
}
]
},
"docs": [
{
"_index": "index",
"_id": "id",
"_source": {
"foo": {
"bar": "baz"
},
"a": {}
}
},
{
"_index": "index",
"_id": "id",
"_source": {
"foo.bar": "baz"
}
}
]
}
- 파이프라인에서
flexible액세스 패턴을 사용합니다. - 다음 필드에서 값을 읽습니다.
foo.bar. - 값을 다음 필드에 씁니다.
a.b.c.d. - 첫 번째 문서 구조는 중첩 JSON 객체를 사용합니다.
- 첫 번째 문서의 루트에는 기존
a필드가 있습니다. - 두 번째 문서는 점으로 구분된 필드 이름을 사용합니다.
{
"docs": [
{
"doc": {
"_id": "id",
"_index": "index",
"_version": "-3",
"_source": {
"foo": {
"bar": "baz"
},
"a": {
"b.c.d": "baz"
}
},
"_ingest": {
"timestamp": "2017-05-04T22:30:03.187Z"
}
}
},
{
"doc": {
"_id": "id",
"_index": "index",
"_version": "-3",
"_source": {
"foo.bar": "baz",
"a.b.c.d": "baz"
},
"_ingest": {
"timestamp": "2017-05-04T22:30:03.188Z"
}
}
}
]
}
- 이
flexible액세스 패턴은 중첩 객체 필드를 지원합니다. 프로세서는foo필드를 찾은 다음bar필드를 찾습니다. - 다음 필드의 값은
foo.bar필드의 값은 점으로 구분된 필드 이름b.c.d에 기록되며 이는a필드 아래에 있습니다. 프로세서는 누락된 필드 이름을 연결하여 키의 접두사로 사용합니다. - 이
flexible액세스 패턴은 점으로 구분된 필드 이름도 지원합니다. 프로세서는 이름이foo인 필드를 찾고, 찾지 못하면 이름이foo.bar. - 다음 필드의 값은
foo.bar필드의 값은 점으로 구분된 필드 이름a.b.c.d에 기록됩니다. 아직 문서에 해당 필드가 하나도 없으므로 점으로 구분된 하나의 필드 이름으로 연결됩니다.
프로세서는 이름으로 다음 메타데이터 필드에 액세스할 수 있습니다.
_index_id_routing_dynamic_templates
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "Set '_routing' to 'geoip.country_iso_code' value",
"field": "_routing",
"value": "{{{geoip.country_iso_code}}}"
}
}
]
}
메타데이터 필드 값에 액세스하려면 Mustache 템플릿 스니펫을 사용합니다. 예를 들어 {{{_routing}}} 는 문서의 라우팅 값을 가져옵니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "Use geo_point dynamic template for address field",
"field": "_dynamic_templates",
"value": {
"address": "geo_point"
}
}
}
]
}
위 set 프로세서는 인덱스 매핑에 아직 필드가 정의되지 않은 경우 ES가 이름이 geo_point 인 동적 템플릿을 다음 필드에 사용하도록 지시합니다. address 이 프로세서는 대량 요청에 이미 정의된 다음 필드의 동적 템플릿을 재정의하지만 address 대량 요청에 정의된 다른 동적 템플릿에는 영향을 주지 않습니다.
문서 ID를 자동 생성 하면 프로세서에서 {{{_id}}} 를 사용할 수 없습니다. Elasticsearch는 자동 생성된 _id 값을 수집 후 할당합니다.
수집 프로세서는 _ingest 키를 사용하여 수집 메타데이터를 추가하고 액세스할 수 있습니다.
소스 및 메타데이터 필드와 달리 Elasticsearch는 기본적으로 수집 메타데이터 필드를 인덱싱하지 않습니다. Elasticsearch는 _ingest 키로 시작하는 소스 필드도 허용합니다. 데이터에 이러한 소스 필드가 포함된 경우 _source._ingest 를 사용하여 액세스합니다.
파이프라인은 기본적으로 _ingest.timestamp 수집 메타데이터 필드만 생성합니다. 이 필드에는 Elasticsearch가 문서의 인덱싱 요청을 받은 시점의 타임스탬프가 포함됩니다. _ingest.timestamp 또는 다른 수집 메타데이터 필드를 인덱싱하려면 set 프로세서를 사용합니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "Index the ingest timestamp as 'event.ingested'",
"field": "event.ingested",
"value": "{{{_ingest.timestamp}}}"
}
}
]
}
파이프라인의 프로세서는 순차적으로 실행됩니다. 기본적으로 프로세서 중 하나가 실패하거나 오류를 만나면 파이프라인 처리가 중지됩니다.
프로세서 실패를 무시하고 파이프라인의 나머지 프로세서를 실행하려면 ignore_failure 을 true.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"rename": {
"description": "Rename 'provider' to 'cloud.provider'",
"field": "provider",
"target_field": "cloud.provider",
"ignore_failure": true
}
}
]
}
다음 on_failure 매개변수로 프로세서 실패 직후 실행할 프로세서 목록을 지정합니다. on_failure 가 지정되면 on_failure 구성이 비어 있어도 Elasticsearch는 이후 파이프라인의 나머지 프로세서를 실행합니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"rename": {
"description": "Rename 'provider' to 'cloud.provider'",
"field": "provider",
"target_field": "cloud.provider",
"on_failure": [
{
"set": {
"description": "Set 'error.message'",
"field": "error.message",
"value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
"override": false
}
}
]
}
}
]
}
중첩 오류 처리를 위해 on_failure 프로세서 목록을 중첩합니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"rename": {
"description": "Rename 'provider' to 'cloud.provider'",
"field": "provider",
"target_field": "cloud.provider",
"on_failure": [
{
"set": {
"description": "Set 'error.message'",
"field": "error.message",
"value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
"override": false,
"on_failure": [
{
"set": {
"description": "Set 'error.message.multi'",
"field": "error.message.multi",
"value": "Document encountered multiple ingest errors",
"override": true
}
}
]
}
}
]
}
}
]
}
파이프라인에 on_failure 도 지정할 수 있습니다. on_failure 값이 없는 프로세서가 실패하면 Elasticsearch는 이 파이프라인 수준 매개변수를 대체 수단으로 사용합니다. Elasticsearch는 파이프라인의 나머지 프로세서를 실행하지 않습니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [ ... ],
"on_failure": [
{
"set": {
"description": "Index document to 'failed-<index>'",
"field": "_index",
"value": "failed-{{{ _index }}}"
}
}
]
}
파이프라인 실패에 관한 추가 정보는 문서 메타데이터 필드 on_failure_message, on_failure_processor_type, on_failure_processor_tag및 on_failure_pipeline에서 확인할 수 있습니다. 이러한 필드는 on_failure 블록 내에서만 액세스할 수 있습니다.
다음 예제는 메타데이터 필드를 사용하여 문서에 파이프라인 실패 정보를 포함합니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [ ... ],
"on_failure": [
{
"set": {
"description": "Record error information",
"field": "error_information",
"value": "Processor {{ _ingest.on_failure_processor_type }} with tag {{ _ingest.on_failure_processor_tag }} in pipeline {{ _ingest.on_failure_pipeline }} failed with message {{ _ingest.on_failure_message }}"
}
}
]
}
각 프로세서는 선택적 if 조건을 지원하며 Painless 스크립트로 작성합니다. 지정하면 프로세서는 if 조건이 true.
if 조건 스크립트는 Painless의 수집 프로세서 컨텍스트에서 실행됩니다. if 조건에서 ctx 값은 읽기 전용입니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"drop": {
"description": "Drop documents with 'network.name' of 'Guest'",
"if": "ctx?.network?.name == 'Guest'"
}
}
]
}
다음 script.painless.regex.enabled 클러스터 설정이 활성화된 경우 if 조건 스크립트에서 정규식을 사용할 수 있습니다. 지원되는 구문은 Painless 정규식.
가능하면 정규식을 사용하지 마세요. 비용이 큰 정규식은 인덱싱 속도를 저하시킬 수 있습니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"set": {
"description": "If 'url.scheme' is 'http', set 'url.insecure' to true",
"if": "ctx.url?.scheme =~ /^http[^s]/",
"field": "url.insecure",
"value": true
}
}
]
}
다음 if 조건은 한 줄의 유효한 JSON으로 지정해야 합니다. 하지만 Kibana 콘솔의 삼중 따옴표 구문을 사용하여 더 큰 스크립트를 작성하고 디버깅할 수 있습니다.
가능하면 복잡하거나 비용이 큰 if 조건 스크립트를 사용하지 마세요. 비용이 큰 조건 스크립트는 인덱싱 속도를 저하시킬 수 있습니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"drop": {
"description": "Drop documents that don't contain 'prod' tag",
"if": """
Collection tags = ctx.tags;
if(tags != null){
for (String tag : tags) {
if (tag.toLowerCase().contains('prod')) {
return false;
}
}
}
return true;
"""
}
}
]
}
또한 저장된 스크립트 를 if 조건으로 지정할 수 있습니다.
PUT _scripts/my-prod-tag-script
{
"script": {
"lang": "painless",
"source": """
Collection tags = ctx.tags;
if(tags != null){
for (String tag : tags) {
if (tag.toLowerCase().contains('prod')) {
return false;
}
}
}
return true;
"""
}
}
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"drop": {
"description": "Drop documents that don't contain 'prod' tag",
"if": { "id": "my-prod-tag-script" }
}
}
]
}
수신 문서에는 객체 필드가 포함되는 경우가 많습니다. 프로세서 스크립트가 상위 객체가 없는 필드에 액세스하려고 하면 Elasticsearch는 NullPointerException를 반환합니다. 이러한 예외를 방지하려면 null 안전 연산자예를 들어 ?.를 사용하고 스크립트를 null 안전하게 작성합니다.
예를 들어 ctx.network?.name.equalsIgnoreCase('Guest') 는 null 안전하지 않습니다. ctx.network?.name 는 null을 반환할 수 있습니다. 스크립트를 'Guest'.equalsIgnoreCase(ctx.network?.name)로 다시 작성하면 null 안전합니다. 그 이유는 Guest 가 항상 null이 아니기 때문입니다.
스크립트를 null 안전하게 다시 작성할 수 없다면 명시적인 null 검사를 포함합니다.
PUT _ingest/pipeline/my-pipeline
{
"processors": [
{
"drop": {
"description": "Drop documents that contain 'network.name' of 'Guest'",
"if": "ctx.network?.name != null && ctx.network.name.contains('Guest')"
}
}
]
}
다음 if 조건을 pipeline 프로세서와 결합하여 기준에 따라 문서에 다른 파이프라인을 적용합니다. 이 파이프라인을 기본 파이프라인 으로 인덱스 템플릿 에서 사용하여 여러 데이터 스트림이나 인덱스를 구성할 수 있습니다.
PUT _ingest/pipeline/one-pipeline-to-rule-them-all
{
"processors": [
{
"pipeline": {
"description": "If 'service.name' is 'apache_httpd', use 'httpd_pipeline'",
"if": "ctx.service?.name == 'apache_httpd'",
"name": "httpd_pipeline"
}
},
{
"pipeline": {
"description": "If 'service.name' is 'syslog', use 'syslog_pipeline'",
"if": "ctx.service?.name == 'syslog'",
"name": "syslog_pipeline"
}
},
{
"fail": {
"description": "If 'service.name' is not 'apache_httpd' or 'syslog', return a failure message",
"if": "ctx.service?.name != 'apache_httpd' && ctx.service?.name != 'syslog'",
"message": "This pipeline requires service.name to be either `syslog` or `apache_httpd`"
}
}
]
}
다음 노드 통계 API를 사용하여 전역 및 파이프라인별 수집 통계를 가져옵니다. 이 통계를 사용하여 가장 자주 실행되거나 처리에 가장 많은 시간을 쓰는 파이프라인을 확인합니다.
GET _nodes/stats/ingest?filter_path=nodes.*.ingest