Elasticsearch Transforms: 여러 상태 업데이트의 총 지속 시간 계산

Elasticsearch Transforms님이

기존 파일을 요약 파일(pivot transforms) 또는

으로 변환

특정 고유 키가 있는 최신 문서를 찾습니다(latest transforms).

이러한 엔티티 중심 뷰는 사용자 비헤이비어나 세션 같은 여러 문서로 구성된 다양한 데이터를 처리하는 데 도움이 됩니다.예를 들어 분포식 시스템에서 세션이나 요청의 지속 시간은 흔히 볼 수 있는 장면이다.다음 게시물은 이터레이션에 반복되는 StackOverflow question을 기반으로 한 것으로, 이를 청사진으로 사용합니다.

샘플 데이터

세 가지 다른 실체가 있는데 그것이 바로 uniqueID, A, B과 C이다.각 eventStart.timestamp 또는 eventStop.timestamp을 사용하여 여러 상태 업데이트를 수행할 수 있습니다.

PUT test/_doc/1
{
  "uniqueID": "A",
  "eventStart": {
    "timestamp": "2020-07-01T13:50:55.000Z"
  }
}
PUT test/_doc/2
{
  "uniqueID": "A",
  "eventStop": {
    "timestamp": "2020-07-01T13:51:00.000Z"
  }
}
PUT test/_doc/3
{
  "uniqueID": "B",
  "eventStart": {
    "timestamp": "2020-07-01T13:52:25.000Z"
  }
}
PUT test/_doc/4
{
  "uniqueID": "B",
  "eventStop": {
    "timestamp": "2020-07-01T13:53:00.000Z"
  }
}
PUT test/_doc/5
{
  "uniqueID": "A",
  "eventStop": {
    "timestamp": "2020-07-01T13:54:55.000Z"
  }
}
PUT test/_doc/6
{
  "uniqueID": "C",
  "eventStart": {
    "timestamp": "2020-07-01T13:54:55.000Z"
  }
}

기본 매핑에 따라 두 개의 date 필드와 keyword 필드는 계산의 다른 지속 시간과 관련이 있습니다.

# Request
GET test/_mapping

# Response
{
  "test" : {
    "mappings" : {
      "properties" : {
        "eventStart" : {
          "properties" : {
            "timestamp" : {
              "type" : "date"
            }
          }
        },
        "eventStop" : {
          "properties" : {
            "timestamp" : {
              "type" : "date"
            }
          }
        },
        "uniqueID" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        }
      }
    }
  }
}

변환 API

다음과 같이 계산됩니다.

조는 uniqueID이다.

은 최초의 eventStart과 마지막 eventStop의 시간 스탬프를 받았다.

은 시간차(초)를 계산합니다.

키바나는 Elasticsearch Transform API의 기초 위에서 하나의 사용자 인터페이스를 제공하여 하나의 작업을 함께 클릭할 수 있지만 본 사례는 Elasticsearch API를 꾸준히 사용하면 따르고 복제하기 쉽다.편리한 API는 preview과 POST _transform/_preview입니다.
그룹을 나누는 첫 번째 단계부터 aggregations 부분은 강제적이기 때문에 상태 업데이트의 수량을 계산합니다.

# Request
POST _transform/_preview
{
  "source": {
    "index": "test"
  },
  "dest": {
    "index": "test_transformed"
  },
  "pivot": {
    "group_by": {
      "id": {
        "terms": {
          "field": "uniqueID.keyword"
        }
      }
    },
    "aggregations": {
      "event_count": {
        "value_count": {
          "field": "_id"
        }
      }
    }
  }
}

# Response
{
  "preview" : [
    {
      "event_count" : 3,
      "id" : "A"
    },
    {
      "event_count" : 2,
      "id" : "B"
    },
    {
      "event_count" : 1,
      "id" : "C"
    }
  ],
  "generated_dest_index" : {
    "mappings" : {
      "_meta" : {
        "_transform" : {
          "transform" : "transform-preview",
          "version" : {
            "created" : "7.13.1"
          },
          "creation_date_in_millis" : 1626059453830
        },
        "created_by" : "transform"
      },
      "properties" : {
        "event_count" : {
          "type" : "long"
        },
        "id" : {
          "type" : "keyword"
        }
      }
    },
    "settings" : {
      "index" : {
        "number_of_shards" : "1",
        "auto_expand_replicas" : "0-1"
      }
    },
    "aliases" : { }
  }
}

최종 결과에 대해 말하자면, 그것은 단지 정확한 aggregation(s):bucket script aggregation을 놓쳤을 뿐이다.

Bucket 스크립트 모음으로 변환

이전 변환을 계속합니다. 이 변환에는 가장 빠른 시작 타임 스탬프, 최신 종료 타임 스탬프 및 둘 사이의 지속 시간이 추가됩니다.

# Request
POST _transform/_preview
{
  "source": {
    "index": "test"
  },
  "dest": {
    "index": "test_transformed"
  },
  "pivot": {
    "group_by": {
      "id": {
        "terms": {
          "field": "uniqueID.keyword"
        }
      }
    },
    "aggregations": {
      "event_count": {
        "value_count": {
          "field": "_id"
        }
      },
      "start": {
        "min": {
          "field": "eventStart.timestamp"
        }
      },
      "stop": {
        "max": {
          "field": "eventStop.timestamp"
        }
      },
      "duration": {
        "bucket_script": {
          "buckets_path": {
            "start": "start.value",
            "stop": "stop.value"
          },
          "script": """
            return (params.stop - params.start)/1000;
          """
        }
      }
    }
  }
}

# Response
{
  "preview" : [
    {
      "duration" : 240.0,
      "stop" : "2020-07-01T13:54:55.000Z",
      "event_count" : 3,
      "start" : "2020-07-01T13:50:55.000Z",
      "id" : "A"
    },
    {
      "duration" : 35.0,
      "stop" : "2020-07-01T13:53:00.000Z",
      "event_count" : 2,
      "start" : "2020-07-01T13:52:25.000Z",
      "id" : "B"
    },
    {
      "stop" : null,
      "event_count" : 1,
      "start" : "2020-07-01T13:54:55.000Z",
      "id" : "C"
    }
  ],
  ...

Painless의 계산은 매우 간단합니다. (params.stop - params.start)/1000:

보다 복잡한 datetime APIs은 필요 없습니다.Elasticsearch의 모든 date은 밀리초 단위로 long으로 저장되기 때문에 간단한 차이만으로도 충분하다.

에서 초로 옮기는 것은 1000의 나눗셈이다.

누락된 종료 시간은 자동으로 처리됩니다.

미리 보기가 아닌 변환 작업을 작성하려면 요청을 다음과 같이 조정해야 합니다.

PUT _transform/test_duration
{
  "description": "Calculate the duration of an event from multiple status updates (based on its uniqueID)",
  "frequency": "1m",
  "source": {
    "index": "test"
  },
  "dest": {
    "index": "test_transformed"
  },
  "pivot": {
    "group_by": {
      "id": {
        "terms": {
          "field": "uniqueID.keyword"
        }
      }
    },
    "aggregations": {
      "event_count": {
        "value_count": {
          "field": "_id"
        }
      },
      "start": {
        "min": {
          "field": "eventStart.timestamp"
        }
      },
      "stop": {
        "max": {
          "field": "eventStop.timestamp"
        }
      },
      "duration": {
        "bucket_script": {
          "buckets_path": {
            "start": "start.value",
            "stop": "stop.value"
          },
          "script": """
            return (params.stop - params.start)/1000;
          """
        }
      }
    }
  }
}

GET _transform/test_duration을 사용하면 전환 작업을 볼 수 있습니다.너는 반드시 explicitly start it과 POST _transform/test_duration/_start을 결합시켜야 한다. 그렇지 않으면 그것은 아무것도 할 수 없다.
마지막으로 stats API은 이 일을 보고 무엇을 하고 있는지 기뻐했습니다.

# Request
GET _transform/test_duration/_stats

# Response
{
  "count" : 1,
  "transforms" : [
    {
      "id" : "test_duration",
      "state" : "stopped",
      "stats" : {
        "pages_processed" : 2,
        "documents_processed" : 6,
        "documents_indexed" : 3,
        "documents_deleted" : 0,
        "trigger_count" : 1,
        "index_time_in_ms" : 41,
        "index_total" : 1,
        "index_failures" : 0,
        "search_time_in_ms" : 20,
        "search_total" : 2,
        "search_failures" : 0,
        "processing_time_in_ms" : 2,
        "processing_total" : 2,
        "delete_time_in_ms" : 0,
        "exponential_avg_checkpoint_duration_ms" : 114.0,
        "exponential_avg_documents_indexed" : 3.0,
        "exponential_avg_documents_processed" : 6.0
      },
      "checkpointing" : {
        "last" : {
          "checkpoint" : 1,
          "timestamp_millis" : 1626063846766
        },
        "changes_last_detected_at" : 1626063846766
      }
    }
  ]
}

Kibana는 같은 결과를 얻었다:
마지막으로 가장 중요하지 않은 것은 다음과 같은 문서가 생성된 것입니다.

# Request
GET test_transformed/_search

# Response
{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 3,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "test_transformed",
        "_type" : "_doc",
        "_id" : "QRRx52klPRvG45a5oLgZ95sAAAAAAAAA",
        "_score" : 1.0,
        "_source" : {
          "duration" : 240.0,
          "stop" : "2020-07-01T13:54:55.000Z",
          "event_count" : 3,
          "start" : "2020-07-01T13:50:55.000Z",
          "id" : "A"
        }
      },
      {
        "_index" : "test_transformed",
        "_type" : "_doc",
        "_id" : "Qq7col5MOHvjTNMiAGonnqAAAAAAAAAA",
        "_score" : 1.0,
        "_source" : {
          "duration" : 35.0,
          "stop" : "2020-07-01T13:53:00.000Z",
          "event_count" : 2,
          "start" : "2020-07-01T13:52:25.000Z",
          "id" : "B"
        }
      },
      {
        "_index" : "test_transformed",
        "_type" : "_doc",
        "_id" : "Q-N5zMGevsgbxCl0WsHH6CIAAAAAAAAA",
        "_score" : 1.0,
        "_source" : {
          "stop" : null,
          "event_count" : 1,
          "start" : "2020-07-01T13:54:55.000Z",
          "id" : "C"
        }
      }
    ]
  }
}

이것이 바로 지속 시간을 계산하는 방법이다🥳

변환 없는 집합

당신은 변환을 해야만 이 결과를 얻을 수 있습니까?아뇨.
몇 가지 수정 사항을 통해 일반 집합을 통해 같은 결과를 얻을 수 있습니다.

# Request
POST test/_search
{
  "size": 0,
  "aggregations": {
    "group_by": {
      "terms": {
        "field": "uniqueID.keyword"
      },
      "aggregations": {
        "start": {
          "min": {
            "field": "eventStart.timestamp"
          }
        },
        "stop": {
          "max": {
            "field": "eventStop.timestamp"
          }
        },
        "duration": {
          "bucket_script": {
            "buckets_path": {
              "start": "start.value",
              "stop": "stop.value"
            },
            "script": """
              return (params.stop - params.start)/1000;
            """
          }
        }
      }
    }
  }
}

# Response
{
  "took" : 9,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 6,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "group_by" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "A",
          "doc_count" : 3,
          "stop" : {
            "value" : 1.593611695E12,
            "value_as_string" : "2020-07-01T13:54:55.000Z"
          },
          "start" : {
            "value" : 1.593611455E12,
            "value_as_string" : "2020-07-01T13:50:55.000Z"
          },
          "duration" : {
            "value" : 240.0
          }
        },
        {
          "key" : "B",
          "doc_count" : 2,
          "stop" : {
            "value" : 1.59361158E12,
            "value_as_string" : "2020-07-01T13:53:00.000Z"
          },
          "start" : {
            "value" : 1.593611545E12,
            "value_as_string" : "2020-07-01T13:52:25.000Z"
          },
          "duration" : {
            "value" : 35.0
          }
        },
        {
          "key" : "C",
          "doc_count" : 1,
          "stop" : {
            "value" : null
          },
          "start" : {
            "value" : 1.593611695E12,
            "value_as_string" : "2020-07-01T13:54:55.000Z"
          }
        }
      ]
    }
  }
}

결과의 구조는 다르지만 결과는 같다. 그리고 몇 가지 보충 설명이 있다.

은 밑바닥 문서를 검색할 필요가 없다. "size": 0은 이렇게 한다.

은 terms aggregation에서 운행하고 기타 sub aggregations에서 운행한다.

doc_count은 상태 업데이트 수량을 자동으로 계산하므로 value_count이 필요하지 않습니다.

과 bucket_script은 같다.

결론

이것은 전환이나 등효 집합의 유용한 청사진이길 바란다.이제 모든 세션이 어떻게 조합되는지, 그리고 어떤 함정 (예를 들어 전환 작업 시작) 을 피해야 하는지 알 수 있습니다. 이것은 즐거운 계산입니다.
이 문서에는 when (not) to use transforms에 대한 설명도 나와 있습니다. 이 설명서는 고전적인 "그것은""토론에 달려 있습니다."
전환은 언제 사용해야 합니까?

대량의 데이터가 있고 완전한 집합이 너무 느리거나 자원이 밀집될 수 있습니다. 특히 대형 대시보드에 자주 조회되거나 표시될 때입니다.

디스크 공간을 절약하기 위해 데이터를 정리하고 각 상태 업데이트가 아닌 실체 중심의 보기에만 관심을 가지고자 하는 경우

, 실체 중심의 문서를 사용하면 더욱 간단합니다. Elasticsearch API나 Kibana를 통해.예를 들어 키바나에서 지속시간의 평균치, 평균치 또는 99번째 백분위수를 계산하면 변환을 사용하는 것은 간단하지만 변환이 없으면 더욱 도전적이다.

언제 전환을 사용하지 말아야 합니까?

, 최신 결과 및 (구성 가능) frequency이 필요한 경우 문제가 될 수 있습니다.

데이터를 거의 읽지 않는다면 전환 작업의 비용은 가치가 없습니다.

런타임 필드를 사용할 수 있습니까?
Runtime fields은 문서에만 필드를 추가하거나 덮어쓰는 또 다른 문제를 해결합니다.런타임 필드 위에서 집합을 실행할 수 있지만, 그 자체는 크로스 문서 기능을 제공하지 않습니다.