사용 방법

NEST Elasticsearch .NET 클라이언트로 문서 색인하기

By
Stuart Cam

서론

NEST Elasticsearch .NET 클라이언트를 사용하여 Elasticsearch에 문서를 인덱싱하는 방법은 여러 가지가 있습니다.

이 블로그 게시물에서는 한 번에 하나의 문서를 색인하는 간단한 방법부터 BulkObservable 헬퍼를 사용하는 고급 방법에 이르기까지 몇 가지 방법을 보여드립니다.

단일 문서

NEST 내에서 문서는 POCO(plain old CLR object)로 모델링되며, 예시는 다음과 같습니다:

public class Person
{
    public int Id { get; set; }
    public 스트링 FirstName { get; set; }
    public 스트링 LastName { get; set; }
}

Elasticsearch에서 단일 문서를 나타내는 이 객체의 인스턴스는 몇 가지 다른 방법을 사용하여 인덱스할 수 있습니다. 다음 인스턴스를 예로 들어 보겠습니다.

var person = new Person
{
    Id = 1,
    FirstName = "Martijn",
    LastName = "Laarman"
};

IndexDocument<T> 및 IndexDocumentAsync<T> 메서드는 기본 매개변수를 사용하여 T 유형의 단일 문서를 색인하는 간단한 방법을 제공합니다. 이 메서드 호출의 결과를 검사하여 색인 작업이 성공했는지 확인할 수 있습니다.

// IIndexResponse 객체를 반환하는 동기식 메서드
var indexResponse = client.IndexDocument(person);
// 대기 가능한 Task<IIndexResponse> 를 반환하는 비동기식 메서드
var indexResponseAsync = await client.IndexDocumentAsync(person);
// 동기식 작업의 결과 검사
if (!indexResponse.IsValid)
{
    // If the request isn't valid, we can take action here
}

IsValid 속성을 사용하여 응답이 기능적으로 유효한지 확인할 수 있습니다. 이는 요청에 문제가 발생했는지 확인하기 위한 단일 지점을 제공하는 NEST 추상화입니다.

문서를 색인할 때 추가 매개변수를 설정해야 하는 경우, fluent 또는 객체 이니셜라이저 구문을 사용할 수 있습니다. 이를 통해 색인 프로세스를 더 세밀하게 제어할 수 있습니다. 아래 예제에서는 “people”이라는 이름의 인덱스에 문서를 색인합니다.

// fluent 구문
var fluentIndexResponse = client.Index(person, i => i.Index("people"));
// 객체 이니셜라이저 구문
var initializerIndexResponse = client.Index(new IndexRequest<Person>(person, "people"));

여러 문서를 색인하는 단순한 접근 방식은 각 반복에서 단일 문서를 색인하도록 루프를 생성하는 것이지만, 이는 대규모 문서 컬렉션에 대해 잘 확장되지 않는 매우 비효율적인 접근 방식입니다.

여러 문서

벌크 API는 여러 문서를 색인하는 데 사용할 수 있습니다. 먼저 색인할 문서 컬렉션을 생성해 보겠습니다:

var people = new []
{
    new Person
    {
        Id = 1,
        FirstName = "Martijn",
        LastName = "Laarman"
    },
    new Person
    {
        Id = 2,
        FirstName = "Stuart",
        LastName = "Cam"
    },
    new Person
    {
        Id = 3,
        FirstName = "Russ",
        LastName = "Cam"
    }
    // snip
};

IndexMany 및 IndexManyAsync 메서드를 사용하여 여러 문서를 각각 동기식 또는 비동기식으로 색인할 수 있습니다. 이러한 메서드는 NEST 클라이언트 전용이며 클라이언트의 Bulk 메서드 및 벌크 API 호출을 래핑하여 많은 문서를 색인하기 위한 편리한 바로가기를 제공합니다.

이러한 메서드는 단일 HTTP 요청으로 모든 문서를 인덱싱하므로, 매우 큰 문서 컬렉션의 경우 컬렉션을 더 작은 여러 배치로 분할하고 여러 Bulk 호출을 실행해야 합니다. 이러한 작업이 필요한 경우, 이 게시물 뒷부분에서 설명하는 BulkAllObservable<T> 헬퍼를 대신 사용하는 것을 고려하십시오.

// IBulkResponse를 반환하는 동기식 메서드
var indexManyResponse = client.IndexMany(people);
if (indexManyResponse.Errors)
{
    // 응답에서 오류를 검사할 수 있음
    foreach (var itemWithError in indexManyResponse.ItemsWithErrors)
    {
        // 오류가 있는 경우 열거하고 검사할 수 있음
        Console.WriteLine("문서 {0} 인덱싱 실패: {1}",
            itemWithError.Id, itemWithError.Error);
    }
}
// 또는 문서를 비동기식으로 인덱싱할 수 있음
var indexManyAsyncResponse = await client.IndexManyAsync(people);

많은 문서를 색인하는 과정에서 더 세밀한 제어가 필요한 경우, Bulk 및 BulkAsync 메서드를 사용하고 디스크립터를 사용하여 벌크 호출을 사용자 지정할 수 있습니다.

위의 IndexMany 메서드와 마찬가지로, 문서는 단일 HTTP 요청으로 _bulk 엔드포인트에 전송됩니다. 이는 HTTP 요청의 전체 크기를 고려해야 함을 의미합니다. 대량의 문서를 색인하려면 BulkAllObservable<T> 헬퍼를 사용하는 것이 좋습니다.

// 오류를 검사할 수 있는 IBulkResponse를 반환합니다
var bulkIndexResponse = client.Bulk(b => b
    .Index("people")
    .IndexMany(people)
);
// 비동기 버전
var asyncBulkIndexResponse = await client.BulkAsync(b => b
    .Index("people")
    .IndexMany(people)
);

BulkAllObservable<T> 헬퍼

BulkAllObservable<T> 헬퍼를 사용하면 재시도, 백오프 또는 배치 메커니즘에 신경 쓸 필요 없이 문서 컬렉션을 색인하는 전체적인 목표에 집중할 수 있습니다.

BulkAll 메서드와 BlockingSubscribeExtensions Wait() 확장 메서드를 사용하여 여러 문서를 색인할 수 있습니다. 이 헬퍼는 색인 실패 시 자동으로 재시도/백오프하고 단일 HTTP 요청에서 색인되는 문서 수를 제어하는 기능을 제공합니다.

다음 예제에서 각 요청은 원본 입력에서 배치 처리된 1000개의 문서를 색인합니다. 문서 수가 많은 경우, 각각 1000개의 문서를 포함하는 많은 HTTP 요청이 발생할 수 있습니다(총 문서 수에 따라 마지막 요청에는 더 적은 문서가 포함될 수 있습니다).

이 헬퍼는 IEnumerable<T> 컬렉션을 지연 열거하여 페이지가 매겨진 데이터베이스 레코드에서 구체화된 문서와 같이 많은 수의 문서를 쉽게 색인할 수 있도록 합니다.

var bulkAllObservable = client.BulkAll(people, b => b
    .Index("people")
    // 재시도 간 대기 시간
    .BackOffTime("30s") 
    // 실패 발생 시 재시도 횟수
    .BackOffRetries(2) 
    // 벌크 작업 완료 후 인덱스 새로 고침
    .RefreshOnCompleted()
    // 동시 벌크 요청 수
    .MaxDegreeOfParallelism(Environment.ProcessorCount)
    // 벌크 요청당 항목 수
    .Size(1000)
)
// 색인하고 최대 15분까지 대기합니다. 
// BulkAll 호출은 비동기식이지만 이는 차단 작업입니다.
.Wait(TimeSpan.FromMinutes(15), next =>
{
    // do something on each response e.g. write number of batches indexed to console
});

BulkAllObservable<T> 헬퍼는 다양한 고급 기능을 제공합니다.

  1. BufferToBulk를 사용하면 벌크 요청이 서버로 전달되기 전에 개별 작업을 사용자 지정할 수 있습니다.
  2. RetryDocumentPredicate를 사용하면 색인에 실패한 문서를 재시도할지 여부를 세밀하게 제어할 수 있습니다.
  3. DroppedDocumentCallback: 재시도 후에도 문서가 인덱싱되지 않는 경우 이 델리게이트가 호출됩니다.
client.BulkAll(people, b => b
    .BufferToBulk((descriptor, list) =>
    {
        // 대량 요청이 발송되기 전에
        // 개별 작업을 사용자 지정합니다.
        foreach (var item in list)
        {
           // index each document into either even-index or odd-index
           descriptor.Index<Person>(bi => bi
              .Index(item.Id % 2 == 0 ? "even-index" : "odd-index")
              .Document(item)
           );
        }
    })
    .RetryDocumentPredicate((item, person) =>
    {
        // decide if a document should be retried in the event of a failure
        return item.Error.Index == "even-index" && person.FirstName == "Martijn";
    })
    .DroppedDocumentCallback((item, person) =>
    {
        // 문서를 인덱싱할 수 없는 경우 이 대리자가 호출됩니다.
        Console.WriteLine($"인덱싱할 수 없음: {item} {person}");
    })
);

수집 노드

Elasticsearch는 수집 요청을 인제스트 노드로 자동으로 재라우팅하므로, 라우팅 정보를 지정하거나 구성할 필요가 없습니다. 그러나 대량의 수집 작업을 수행하고 전용 인제스트 노드가 있는 경우, 클러스터 내에서 불필요한 홉(hop)을 방지하기 위해 인덱스 요청을 이러한 노드로 직접 보내는 것이 합리적입니다.

이를 달성하는 가장 간단한 방법은 전용 "색인" 클라이언트 인스턴스를 생성하고 이를 색인 요청에 사용하는 것입니다.

// 인제스트 노드 목록
var pool = new StaticConnectionPool(new []
{
    new Uri("http://ingestnode1:9200"),
    new Uri("http://ingestnode2:9200"),
    new Uri("http://ingestnode3:9200")
});
var settings = new ConnectionSettings(pool);
var indexingClient = new ElasticClient(settings);

복잡한 클러스터 구성에서는 스니핑 연결 풀을 Node 조건자(node predicate)와 함께 사용하여 인제스트 기능이 있는 Node를 필터링하는 것이 더 쉬울 수 있습니다. 이를 통해 클라이언트를 재구성할 필요 없이 클러스터를 사용자 지정할 수 있습니다.

// 클러스터 Node 목록
var pool = new SniffingConnectionPool(new []
{
    new Uri("http://node1:9200"),
    new Uri("http://node2:9200"),
    new Uri("http://node3:9200")
});
// 인제스트 기능이 있는 Node만 선택하기 위한 조건자
var settings = new ConnectionSettings(pool).NodePredicate(n => n.IngestEnabled);
var indexingClient = new ElasticClient(settings);

수집 파이프라인

추가 정보를 포함하도록 Person 유형을 수정해 보겠습니다.

public class Person
{
    public int Id { get; set; }
    public 스트링 FirstName { get; set; }
    public 스트링 LastName { get; set; }
    public 스트링 IpAddress { get; set; }
    public GeoIp GeoIp { get; set; }
}
public class GeoIp
{
    public 스트링 CityName { get; set; }
    public 스트링 ContinentName { get; set; }
    public 스트링 CountryIsoCode { get; set; }
    public GeoLocation Location { get; set; }
    public 스트링 RegionName { get; set; }
}

수집 파이프라인을 생성하여 인덱싱되기 전에 들어오는 값을 조작할 수 있습니다. 애플리케이션이 항상 성(surname)을 대문자로 표기하고, 이니셜을 별도의 필드로 인덱싱할 것으로 예상한다고 가정해 보겠습니다. 또한 사람이 읽을 수 있는 위치 정보로 변환하려는 IP 주소도 있습니다.

사용자 정의 매핑과 인제스트 파이프라인을 생성하여 이 요구 사항을 달성할 수 있습니다. 그러면 추가 변경 없이 새로운 Person 유형을 사용할 수 있습니다.

먼저 인덱스와 사용자 지정 매핑을 생성합니다:

client.CreateIndex("people", c => c
    .Mappings(ms => ms
        .Map<Person>(p => p
            //유형에서 매핑을 자동으로 생성
            .AutoMap()
            //AutoMap()에서 추론된 매핑을 재정의 
            .Properties(props => props
                // 이니셜을 저장할 추가 필드 생성
                .Keyword(t => t.Name("initials"))
                 //필드를 IP 주소 유형으로 매핑
                .Ip(t => t.Name(dv => dv.IpAddress))
                // GeoIp를 객체로 매핑
                .Object<GeoIp>(t => t.Name(dv => dv.GeoIp))
            )
        )
    )
);

다음으로 버전 6.7부터 번들로 제공되는 인제스트-geoip 플러그인을 활용하여 인제스트 파이프라인을 생성하겠습니다.

client.PutPipeline("person-pipeline", p => p
    .Processors(ps => ps
        //성(lastname)을 대문자로 변환
        .Uppercase<Person>(s => s
            .Field(t => t.LastName)
        )
        // Painless 스크립트를 사용하여 새 필드 채우기
        .Script(s => s
            .Lang("painless") 
            .Source("ctx.initials = ctx.firstName.substring(0,1) + ctx.lastName.substring(0,1)")
        )
        // 인제스트-geoip 플러그인을 사용하여 제공된 IP 주소에서 GeoIp 객체 보강
        .GeoIp<Person>(s => s 
            .Field(i => i.IpAddress)
            .TargetField(i => i.GeoIp)
        )
    )
);

이제 이 새로운 인덱스와 인제스트 파이프라인을 사용하여 Person 인스턴스를 인덱싱해 보겠습니다.

var person = new Person
{
    Id = 1,
    FirstName = "Martijn",
    LastName = "Laarman",
    IpAddress = "139.130.4.5"
};
// 생성된 파이프라인을 사용하여 문서 인덱싱
var indexResponse = client.Index(person, p => p
    .Index("people")
    .Pipeline("person-pipeline")
);

이제 검색하면 보강된 값이 포함된 인덱스 문서가 표시됩니다.

{
  "took": 5,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 1,
    "max_score": 1,
    "hits": [
      {
        "_index": "people",
        "_type": "person",
        "_id": "1",
        "_score": 1,
        "_source": {
          "firstName": "Martijn",
          "lastName": "LAARMAN",
          "initials": "ML",
          "geoIp": {
            "continent_name": "Oceania",
            "region_iso_code": "AU-NSW",
            "city_name": "Sydney",
            "country_iso_code": "AU",
            "region_name": "New South Wales",
            "location": {
              "lon": 151.2167,
              "lat": -33.7333
            }
          },
          "ipAddress": "139.130.4.5",
          "id": 1
        }
      }
    ]
  }
}

파이프라인이 지정되면 색인 시 문서 보강에 따른 오버헤드가 추가됩니다. 위 예시에서는 대문자 변환 및 Painless 스크립트 실행이 이에 해당합니다.

대규모 벌크 요청의 경우 예외를 방지하기 위해 기본 색인 타임아웃을 늘리는 것이 좋습니다.

client.Bulk(b => b
    .Index("people")
    .파이프라인("person-pipeline")
    //Elasticsearch 서버 측 타임아웃 증가
    .Timeout("5m") 
    .IndexMany<Person>(people)
    .RequestConfiguration(rc => rc
        // 요청을 중단하기 전 클라이언트의 HTTP 요청 타임아웃 증가
        .RequestTimeout(TimeSpan.FromMinutes(5)) 
    )
);

요약

이 블로그 게시물에서는 단일 문서를 색인하는 간단한 사례부터 인제스트 파이프라인을 사용하여 여러 문서를 대량으로 색인하는 방법까지 다루었습니다.

사용자의 클러스터에서 직접 시도해 보거나 Elastic Cloud 의 Elasticsearch Service 14일 무료 체험판을 시작해 보십시오. 문제가 발생하거나 궁금한 점이 있으면 Discuss 포럼에서 문의해 주시기 바랍니다.

NEST Elasticsearch .NET 클라이언트를 사용한 색인 작업에 대한 전체 설명서는 당사 설명서를 참조하십시오.