수집 스펙 작성하기
수집 스펙 작성하기 (Write an ingestion spec)
수집 스펙(ingestion spec)을 정의하는 과정을 단계별로 안내하고, 핵심 고려 사항과 지침을 짚어 드릴게요. Druid의 dataSchema 를 비우지 않은 상태에서 시작해 필드를 하나씩 채워 가면서 네이티브 배치 수집 태스크를 완성해 봐요.
출처: 문서
본문
이 튜토리얼에서는 단일 머신 퀵스타트 (single-machine quickstart)에 설명된 대로 Apache Druid를 이미 다운로드하고 로컬 머신에서 실행 중이라고 가정해요.
또한 Tutorial: Loading a file, Tutorial: Querying data, Tutorial: Rollup을 완료했으면 도움이 돼요.
예제 데이터 (Example data)
다음과 같은 네트워크 흐름(network flow) 데이터가 있다고 가정해 볼게요:
srcIP: 보내는 쪽의 IP 주소srcPort: 보내는 쪽의 포트dstIP: 받는 쪽의 IP 주소dstPort: 받는 쪽의 포트protocol: IP 프로토콜 번호packets: 전송된 패킷 수bytes: 전송된 바이트 수cost: 트래픽을 보내는 비용
{"ts":"2018-01-01T01:01:35Z","srcIP":"1.1.1.1", "dstIP":"2.2.2.2", "srcPort":2000, "dstPort":3000, "protocol": 6, "packets":10, "bytes":1000, "cost": 1.4}
{"ts":"2018-01-01T01:01:51Z","srcIP":"1.1.1.1", "dstIP":"2.2.2.2", "srcPort":2000, "dstPort":3000, "protocol": 6, "packets":20, "bytes":2000, "cost": 3.1}
{"ts":"2018-01-01T01:01:59Z","srcIP":"1.1.1.1", "dstIP":"2.2.2.2", "srcPort":2000, "dstPort":3000, "protocol": 6, "packets":30, "bytes":3000, "cost": 0.4}
{"ts":"2018-01-01T01:02:14Z","srcIP":"1.1.1.1", "dstIP":"2.2.2.2", "srcPort":5000, "dstPort":7000, "protocol": 6, "packets":40, "bytes":4000, "cost": 7.9}
{"ts":"2018-01-01T01:02:29Z","srcIP":"1.1.1.1", "dstIP":"2.2.2.2", "srcPort":5000, "dstPort":7000, "protocol": 6, "packets":50, "bytes":5000, "cost": 10.2}
{"ts":"2018-01-01T01:03:29Z","srcIP":"1.1.1.1", "dstIP":"2.2.2.2", "srcPort":5000, "dstPort":7000, "protocol": 6, "packets":60, "bytes":6000, "cost": 4.3}
{"ts":"2018-01-01T02:33:14Z","srcIP":"7.7.7.7", "dstIP":"8.8.8.8", "srcPort":4000, "dstPort":5000, "protocol": 17, "packets":100, "bytes":10000, "cost": 22.4}
{"ts":"2018-01-01T02:33:45Z","srcIP":"7.7.7.7", "dstIP":"8.8.8.8", "srcPort":4000, "dstPort":5000, "protocol": 17, "packets":200, "bytes":20000, "cost": 34.5}
{"ts":"2018-01-01T02:35:45Z","srcIP":"7.7.7.7", "dstIP":"8.8.8.8", "srcPort":4000, "dstPort":5000, "protocol": 17, "packets":300, "bytes":30000, "cost": 46.3}
위 JSON 내용을 quickstart/ 에 ingestion-tutorial-data.json 이라는 파일로 저장해 주세요.
이 데이터를 로드할 수 있는 수집 스펙을 정의하는 과정을 함께 살펴볼게요.
이 튜토리얼에서는 네이티브 배치 인덱싱 태스크(native batch indexing task)를 사용할 거예요. 다른 태스크 유형을 사용하면 수집 스펙의 일부 측면이 달라지는데, 튜토리얼이 그런 부분을 짚어 줄 거예요.
스키마 정의하기 (Defining the schema)
Druid 수집 스펙의 핵심 요소는 dataSchema 예요. dataSchema 는 입력 데이터를 Druid에 저장될 컬럼 집합으로 어떻게 파싱할지 정의해요.
빈 dataSchema 로 시작해서 튜토리얼을 진행하면서 필드를 하나씩 추가해 볼게요.
quickstart/ 에 ingestion-tutorial-index.json 이라는 새 파일을 만들고 다음 내용을 넣어 주세요:
"dataSchema" : {}
튜토리얼을 진행하면서 이 수집 스펙을 계속 수정할 거예요.
데이터소스 이름 (Datasource name)
데이터소스 이름은 dataSchema 의 dataSource 파라미터로 지정해요.
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
}
튜토리얼 데이터소스를 ingestion-tutorial 이라고 부르기로 해요.
시간 컬럼 (Time column)
dataSchema 는 입력 데이터에서 기본 타임스탬프 필드를 어떻게 추출할지 알아야 해요.
입력 데이터의 타임스탬프 컬럼은 ts 라는 이름이고 ISO 8601 타임스탬프를 담고 있어요. 이 정보로 timestampSpec 을 dataSchema 에 추가해 볼게요:
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
}
}
컬럼 타입 (Column types)
시간 컬럼을 정의했으니, 이제 다른 컬럼들의 정의를 살펴볼게요.
Druid는 다음 컬럼 타입을 지원해요: String, Long, Float, Double. 이 타입들이 다음 섹션에서 어떻게 쓰이는지 볼 거예요.
다른 비-시간 컬럼을 어떻게 정의할지로 넘어가기 전에, 먼저 rollup에 대해 이야기해 볼게요.
Rollup
데이터를 수집할 때는 rollup을 사용할지 여부를 고려해야 해요.
- rollup이 활성화되면 입력 컬럼을 두 범주, 즉 "dimensions"와 "metrics"로 나눠야 해요. "dimensions"는 rollup의 그룹핑 컬럼이고, "metrics"는 집계될 컬럼이에요.
- rollup이 비활성화되면 모든 컬럼이 "dimensions"로 취급되고 사전 집계가 일어나지 않아요.
이 튜토리얼에서는 rollup을 활성화할 거예요. 이는 dataSchema 의 granularitySpec 으로 지정해요.
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"granularitySpec" : {
"rollup" : true
}
}
dimensions와 metrics 선택하기 (Choosing dimensions and metrics)
이 예제 데이터셋에 대해 다음은 "dimensions"와 "metrics"의 합리적인 분할이에요:
- Dimensions: srcIP, srcPort, dstIP, dstPort, protocol
- Metrics: packets, bytes, cost
여기서 dimensions는 IP 트래픽의 단방향 흐름(unidirectional flow)을 식별하는 속성 그룹이고, metrics는 dimension 그룹이 지정한 IP 트래픽 흐름에 대한 사실(facts)을 나타내요.
이 dimensions와 metrics를 수집 스펙 안에서 어떻게 정의하는지 살펴볼게요.
Dimensions
Dimensions는 dataSchema 안의 dimensionsSpec 으로 지정해요.
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" }
]
},
"granularitySpec" : {
"rollup" : true
}
}
각 dimension은 name 과 type 을 가지며, type 은 "long", "float", "double", 또는 "string" 일 수 있어요.
srcIP 가 "string" dimension이라는 점에 주목해 주세요. string dimension의 경우 "string"이 기본 dimension 타입이므로 dimension 이름만 지정하면 충분해요.
또한 protocol 은 입력 데이터에서 숫자 값이지만 "string" 컬럼으로 수집하고 있다는 점도 주목하세요. Druid는 수집 중에 입력 longs를 문자열로 강제 변환(coerce)해요.
문자열 vs 숫자 (Strings vs. Numerics)
숫자 입력을 숫자 dimension으로 수집할까요, 아니면 문자열 dimension으로 수집할까요?
숫자 dimension은 String dimension에 비해 다음과 같은 장단점이 있어요:
- 장점: 숫자 표현이 디스크의 컬럼 크기를 더 작게 만들 수 있고, 컬럼에서 값을 읽을 때 처리 오버헤드가 더 낮아요.
- 단점: 숫자 dimension에는 인덱스가 없어서, 필터링이 (비트맵 인덱스가 있는) 동일한 String dimension을 필터링하는 것보다 느린 경우가 많아요.
Metrics
Metrics는 dataSchema 안의 metricsSpec 으로 지정해요:
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" }
]
},
"metricsSpec" : [
{ "type" : "count", "name" : "count" },
{ "type" : "longSum", "name" : "packets", "fieldName" : "packets" },
{ "type" : "longSum", "name" : "bytes", "fieldName" : "bytes" },
{ "type" : "doubleSum", "name" : "cost", "fieldName" : "cost" }
],
"granularitySpec" : {
"rollup" : true
}
}
metric을 정의할 때는 rollup 동안 그 컬럼에 어떤 유형의 집계를 수행할지 지정해야 해요.
여기서는 두 개의 long metric 컬럼인 packets 와 bytes 에 long sum 집계를, cost 컬럼에는 double sum 집계를 정의했어요.
또한 count aggregator도 정의했다는 점에 주목하세요. count aggregator는 원본 입력 데이터에서 몇 개의 행이 최종 수집 데이터의 "rolled up" 행에 기여했는지를 추적해요.
Rollup 없이 (No rollup)
rollup을 사용하지 않는다면 모든 컬럼이 dimensionsSpec 에 지정돼야 해요. 예를 들면:
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" },
{ "name" : "packets", "type" : "long" },
{ "name" : "bytes", "type" : "long" },
{ "name" : "srcPort", "type" : "double" }
]
},
Granularities 정의하기 (Define granularities)
이 시점에서 dataSchema 안의 dimensionsSpec 과 metricsSpec 정의를 마쳤으니 수집 스펙 작성이 거의 끝났어요.
granularitySpec 에 추가로 설정할 속성 몇 가지가 있어요:
- granularitySpec의 유형:
uniformgranularity spec은 균일한 구간 크기의 세그먼트를 정의해요. 예를 들어 모든 세그먼트가 한 시간 분량의 데이터를 담아요. - 세그먼트 granularity: 단일 세그먼트가 어느 크기의 시간 구간 데이터를 담아야 할까요? 예:
DAY,WEEK - 시간 컬럼의 타임스탬프를 버킷으로 묶는 granularity (
queryGranularity)
세그먼트 granularity (Segment granularity)
세그먼트 granularity는 granularitySpec 의 segmentGranularity 속성으로 구성해요. 이 튜토리얼에서는 시간별 세그먼트를 만들 거예요:
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" }
]
},
"metricsSpec" : [
{ "type" : "count", "name" : "count" },
{ "type" : "longSum", "name" : "packets", "fieldName" : "packets" },
{ "type" : "longSum", "name" : "bytes", "fieldName" : "bytes" },
{ "type" : "doubleSum", "name" : "cost", "fieldName" : "cost" }
],
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "HOUR",
"rollup" : true
}
}
입력 데이터에는 서로 다른 두 시간의 이벤트가 있으므로, 이 태스크는 두 개의 세그먼트를 생성해요.
쿼리 granularity (Query granularity)
쿼리 granularity는 granularitySpec 의 queryGranularity 속성으로 구성해요. 이 튜토리얼에서는 분(minute) 단위 granularity를 사용해 볼게요:
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" }
]
},
"metricsSpec" : [
{ "type" : "count", "name" : "count" },
{ "type" : "longSum", "name" : "packets", "fieldName" : "packets" },
{ "type" : "longSum", "name" : "bytes", "fieldName" : "bytes" },
{ "type" : "doubleSum", "name" : "cost", "fieldName" : "cost" }
],
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "HOUR",
"queryGranularity" : "MINUTE",
"rollup" : true
}
}
쿼리 granularity의 효과를 보려면 원본 입력 데이터의 다음 행을 살펴볼게요:
{"ts":"2018-01-01T01:03:29Z","srcIP":"1.1.1.1", "dstIP":"2.2.2.2", "srcPort":5000, "dstPort":7000, "protocol": 6, "packets":60, "bytes":6000, "cost": 4.3}
이 행이 분 단위 queryGranularity로 수집되면, Druid는 행의 타임스탬프를 분 버킷으로 내림(floor)해요:
{"ts":"2018-01-01T01:03:00Z","srcIP":"1.1.1.1", "dstIP":"2.2.2.2", "srcPort":5000, "dstPort":7000, "protocol": 6, "packets":60, "bytes":6000, "cost": 4.3}
구간 정의하기 (배치 전용, Define an interval)
배치 태스크에서는 시간 구간(interval)을 정의해야 해요. 시간 구간 밖의 타임스탬프를 가진 입력 행은 수집되지 않아요.
구간 역시 granularitySpec 에서 지정해요:
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" }
]
},
"metricsSpec" : [
{ "type" : "count", "name" : "count" },
{ "type" : "longSum", "name" : "packets", "fieldName" : "packets" },
{ "type" : "longSum", "name" : "bytes", "fieldName" : "bytes" },
{ "type" : "doubleSum", "name" : "cost", "fieldName" : "cost" }
],
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "HOUR",
"queryGranularity" : "MINUTE",
"intervals" : ["2018-01-01/2018-01-02"],
"rollup" : true
}
}
태스크 유형 정의하기 (Define the task type)
이제 dataSchema 정의를 마쳤어요. 남은 단계는 만든 dataSchema 를 수집 태스크 스펙에 넣고 입력 소스를 지정하는 거예요.
dataSchema 는 모든 태스크 유형에서 공유되지만, 각 태스크 유형은 고유한 스펙 형식을 가져요. 이 튜토리얼에서는 네이티브 배치 수집 태스크를 사용할 거예요:
{
"type" : "index_parallel",
"spec" : {
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" }
]
},
"metricsSpec" : [
{ "type" : "count", "name" : "count" },
{ "type" : "longSum", "name" : "packets", "fieldName" : "packets" },
{ "type" : "longSum", "name" : "bytes", "fieldName" : "bytes" },
{ "type" : "doubleSum", "name" : "cost", "fieldName" : "cost" }
],
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "HOUR",
"queryGranularity" : "MINUTE",
"intervals" : ["2018-01-01/2018-01-02"],
"rollup" : true
}
}
}
}
입력 소스 정의하기 (Define the input source)
이제 입력 소스를 정의해 볼게요. 입력 소스는 ioConfig 객체에 지정돼요. 각 태스크 유형은 고유한 유형의 ioConfig 를 가져요. 입력 데이터를 읽으려면 inputSource 를 지정해야 해요. 앞서 저장한 netflow 데이터 예제는 로컬 파일에서 읽어야 하는데, 아래와 같이 구성해요:
"ioConfig" : {
"type" : "index_parallel",
"inputSource" : {
"type" : "local",
"baseDir" : "quickstart/",
"filter" : "ingestion-tutorial-data.json"
}
}
데이터 형식 정의하기 (Define the format of the data)
입력 데이터가 JSON 문자열로 표현되므로, inputFormat 을 json 형식으로 사용할 거예요:
"ioConfig" : {
"type" : "index_parallel",
"inputSource" : {
"type" : "local",
"baseDir" : "quickstart/",
"filter" : "ingestion-tutorial-data.json"
},
"inputFormat" : {
"type" : "json"
}
}
{
"type" : "index_parallel",
"spec" : {
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" }
]
},
"metricsSpec" : [
{ "type" : "count", "name" : "count" },
{ "type" : "longSum", "name" : "packets", "fieldName" : "packets" },
{ "type" : "longSum", "name" : "bytes", "fieldName" : "bytes" },
{ "type" : "doubleSum", "name" : "cost", "fieldName" : "cost" }
],
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "HOUR",
"queryGranularity" : "MINUTE",
"intervals" : ["2018-01-01/2018-01-02"],
"rollup" : true
}
},
"ioConfig" : {
"type" : "index_parallel",
"inputSource" : {
"type" : "local",
"baseDir" : "quickstart/",
"filter" : "ingestion-tutorial-data.json"
},
"inputFormat" : {
"type" : "json"
}
}
}
}
추가 튜닝 (Additional tuning)
각 수집 태스크에는 사용자가 다양한 수집 파라미터를 튜닝할 수 있는 tuningConfig 섹션이 있어요.
예를 들어 네이티브 배치 수집 태스크의 목표 세그먼트 크기를 설정하는 tuningConfig 를 추가해 볼게요:
"tuningConfig" : {
"type" : "index_parallel",
"partitionsSpec": {
"type": "dynamic",
"maxRowsPerSegment" : 5000000
}
}
각 수집 태스크는 고유한 유형의 tuningConfig 를 가진다는 점에 주목해 주세요.
최종 스펙 (Final spec)
수집 스펙 정의를 마쳤으니, 이제 다음과 같이 보일 거예요:
{
"type" : "index_parallel",
"spec" : {
"dataSchema" : {
"dataSource" : "ingestion-tutorial",
"timestampSpec" : {
"format" : "iso",
"column" : "ts"
},
"dimensionsSpec" : {
"dimensions": [
"srcIP",
{ "name" : "srcPort", "type" : "long" },
{ "name" : "dstIP", "type" : "string" },
{ "name" : "dstPort", "type" : "long" },
{ "name" : "protocol", "type" : "string" }
]
},
"metricsSpec" : [
{ "type" : "count", "name" : "count" },
{ "type" : "longSum", "name" : "packets", "fieldName" : "packets" },
{ "type" : "longSum", "name" : "bytes", "fieldName" : "bytes" },
{ "type" : "doubleSum", "name" : "cost", "fieldName" : "cost" }
],
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "HOUR",
"queryGranularity" : "MINUTE",
"intervals" : ["2018-01-01/2018-01-02"],
"rollup" : true
}
},
"ioConfig" : {
"type" : "index_parallel",
"inputSource" : {
"type" : "local",
"baseDir" : "quickstart/",
"filter" : "ingestion-tutorial-data.json"
},
"inputFormat" : {
"type" : "json"
}
},
"tuningConfig" : {
"type" : "index_parallel",
"partitionsSpec": {
"type": "dynamic",
"maxRowsPerSegment" : 5000000
}
}
}
}
태스크 제출하고 데이터 쿼리하기 (Submit the task and query the data)
apache-druid-37.0.0 패키지 루트에서 다음 명령을 실행해 주세요:
bin/post-index-task --file quickstart/ingestion-tutorial-index.json --url http://localhost:8081
스크립트가 완료된 뒤 데이터를 쿼리해 볼게요.
웹 콘솔의 Query 뷰에서 새 탭을 열어 주세요. 다음 쿼리를 실행해 수집된 데이터를 확인해 주세요:
select * from "ingestion-tutorial"
다음과 같은 결과가 반환돼요:
__time |
bytes |
cost |
count |
dstIP |
dstPort |
packets |
protocol |
srcIP |
srcPort |
|---|---|---|---|---|---|---|---|---|---|
| 2018-01-01T01:01:00.000Z | 6000 | 4.9 | 3 | 2.2.2.2 | 3000 | 60 | 6 | 1.1.1.1 | 2000 |
| 2018-01-01T01:02:00.000Z | 9000 | 18.1 | 2 | 2.2.2.2 | 7000 | 90 | 6 | 1.1.1.1 | 5000 |
| 2018-01-01T01:03:00.000Z | 6000 | 4.3 | 1 | 2.2.2.2 | 7000 | 60 | 6 | 1.1.1.1 | 5000 |
| 2018-01-01T02:33:00.000Z | 30000 | 56.9 | 2 | 8.8.8.8 | 5000 | 300 | 17 | 7.7.7.7 | 4000 |
| 2018-01-01T02:35:00.000Z | 30000 | 46.3 | 1 | 8.8.8.8 | 5000 | 300 | 17 | 7.7.7.7 | 4000 |
더 알아보기 (Learn more)
- 수집 (Ingestion overview) — 다양한 수집 방법 소개.
- 네이티브 배치 수집 (Native batch ingestion) — 배치 수집 태스크 상세 참조.
- 스키마 모델 (Schema model) — dimensions와 metrics의 정의.