Protobuf

Protobuf

이 Apache Druid 확장은 Druid가 Protobuf 데이터 형식을 수집하고 이해할 수 있게 해 줘요. 확장 로드 목록에 반드시 druid-protobuf-extensions를 포함해야 합니다.

출처: 문서

본문

druid-protobuf-extensions는 스트림 수집을 위한 Protobuf InputFormat을 제공해요. 자세한 내용은 해당 문서를 참고하세요.

예시: Kafka에서 Protobuf 메시지 로드하기

이 예시는 Kafka에서 Protobuf 메시지를 로드하는 방법을 보여줘요. 먼저 Load from Kafka 튜토리얼을 읽고, 자세한 내용은 Kafka Indexing Service 문서를 참고해 주세요.

이 예시에서 사용하는 파일들은 Druid 디렉토리의 ./examples/quickstart/protobuf에 있습니다.

이 예시에서:

  • Kafka broker 호스트는 localhost:9092
  • Kafka 토픽은 metrics_pb
  • Datasource 이름은 metrics-protobuf

다음은 예시에서 사용하는 'metrics' 데이터 스키마의 JSON 예시예요.

{  "unit": "milliseconds",  "http_method": "GET",  "value": 44,  "timestamp": "2017-04-06T02:36:22Z",  "http_code": "200",  "page": "/",  "metricType": "request/latency",  "server": "www1.example.com"}

Proto 파일

우리 'metrics' 데이터셋에 해당하는 proto 파일은 다음과 같아요. Protobuf inputFormat을 proto 파일과 함께 쓰거나 Confluent Schema Registry로 사용할 수 있습니다.

syntax = "proto3";message Metrics {  string unit = 1;  string http_method = 2;  int32 value = 3;  string timestamp = 4;  string http_code = 5;  string page = 6;  string metricType = 7;  string server = 8;}

descriptor 파일을 쓸 때

다음으로 protoc Protobuf 컴파일러를 사용해서 descriptor 파일을 생성하고 metrics.desc로 저장해요. descriptor 파일은 클래스패스에 있거나 URL로 접근 가능해야 합니다. 이 예시에서는 descriptor 파일을 /tmp/metrics.desc에 저장했는데, 이 파일은 예시 파일에도 포함돼 있어요. Druid 설치 디렉토리에서:

protoc -o /tmp/metrics.desc ./quickstart/protobuf/metrics.proto

Schema Registry를 쓸 때

Schema Registry 버전이 5.5보다 이후인지 확인하세요. 그다음 스키마를 레지스트리에 추가하기 위해 스키마를 게시할 수 있어요.

POST /subjects/test/versions HTTP/1.1Host: schemaregistry.example1.comAccept: application/vnd.schemaregistry.v1+json, application/vnd.schemaregistry+json, application/json{    "schemaType": "PROTOBUF",    "schema": "syntax = \"proto3\";\nmessage Metrics {\n  string unit = 1;\n  string http_method = 2;\n  int32 value = 3;\n string timestamp = 4;\n string http_code = 5;\n string page = 6;\n string metricType = 7;\n string server = 8;\n}\n"}

이 기능은 Druid 배포판에 포함되지 않은 Confluent의 Protobuf provider를 사용해요. 따라서 별도로 설치해야 합니다. 이 provider와 그 의존성은 Confluent 저장소와 Maven Central에서 다음 위치로 받을 수 있어요.

이 파일들을 배포판 루트 디렉토리 아래 extensions-core/protobuf-extensions 폴더 안에 복사하거나 심링크하세요.

Kafka Supervisor 만들기

아래는 Overlord에 제출할 완전한 Supervisor 스펙 JSON이에요. 성공적인 수집을 위해 이 키들이 올바르게 구성되어 있는지 확인하세요.

descriptor 파일을 쓸 때

중요한 supervisor 속성

  • protoBytesDecoder.descriptor — descriptor 파일 URL
  • protoBytesDecoder.protoMessageType — proto 정의에서 가져온 값
  • protoBytesDecoder.type — file로 설정, descriptor 파일로 Protobuf 파일을 디코딩함을 나타냄
  • inputFormat — type을 protobuf로 설정
{    "type": "kafka",    "spec": {        "dataSchema": {            "dataSource": "metrics-protobuf",            "timestampSpec": {                "column": "timestamp",                "format": "auto"            },            "dimensionsSpec": {                "dimensions": [                    "unit",                    "http_method",                    "http_code",                    "page",                    "metricType",                    "server"                ],                "dimensionExclusions": [                    "timestamp",                    "value"                ]            },            "metricsSpec": [                {                    "name": "count",                    "type": "count"                },                {                    "name": "value_sum",                    "fieldName": "value",                    "type": "doubleSum"                },                {                    "name": "value_min",                    "fieldName": "value",                    "type": "doubleMin"                },                {                    "name": "value_max",                    "fieldName": "value",                    "type": "doubleMax"                }            ],            "granularitySpec": {                "type": "uniform",                "segmentGranularity": "HOUR",                "queryGranularity": "NONE"            }        },        "tuningConfig": {            "type": "kafka",            "maxRowsPerSegment": 5000000        },        "ioConfig": {            "topic": "metrics_pb",            "consumerProperties": {                "bootstrap.servers": "localhost:9092"            },            "inputFormat": {                "type": "protobuf",                "protoBytesDecoder": {                    "type": "file",                    "descriptor": "file:///tmp/metrics.desc",                    "protoMessageType": "Metrics"                },                "flattenSpec": {                    "useFieldDiscovery": true                },                "binaryAsString": false            },            "taskCount": 1,            "replicas": 1,            "taskDuration": "PT1H",            "type": "kafka"        }    }}

Schema Registry를 쓸 때

중요한 supervisor 속성

  • protoBytesDecoder.url — 단일 인스턴스용 Schema Registry URL
  • protoBytesDecoder.urls — 다중 인스턴스용 Schema Registry URL들
  • protoBytesDecoder.capacity — 스키마 레지스트리에 캐시된 스키마 용량(capacity)
  • protoBytesDecoder.config — Schema Registry용으로 구성된 추가 설정을 보내는 속성
  • protoBytesDecoder.headers — Schema Registry에 보낼 헤더
  • protoBytesDecoder.type — schema_registry로 설정, 스키마 레지스트리로 Protobuf 파일을 디코딩함을 나타냄
{  "urls": ["http://schemaregistry.example1.com:8081","http://schemaregistry.example2.com:8081"],  "type": "schema_registry",  "capacity": 100,  "config" : {       "basic.auth.credentials.source": "USER_INFO",       "basic.auth.user.info": "fred:letmein",       "schema.registry.ssl.truststore.location": "/some/secrets/kafka.client.truststore.jks",       "schema.registry.ssl.truststore.password": "<password>",       "schema.registry.ssl.keystore.location": "/some/secrets/kafka.client.keystore.jks",       "schema.registry.ssl.keystore.password": "<password>",       "schema.registry.ssl.key.password": "<password>",         ...   },  "headers": {      "traceID" : "b29c5de2-0db4-490b-b421",      "timeStamp" : "1577191871865",      ...  }}

Kafka에 Protobuf 메시지 추가하기

필요하다면 Kafka 설치 디렉토리에서 다음 명령을 실행해서 Kafka 토픽을 만들 수 있어요.

./bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic metrics_pb

이 예시 스크립트는 protobuf와 kafka-python 모듈이 필요해요. 토픽이 준비되면 Druid 설치 디렉토리에서 다음 명령을 실행해 메시지를 삽입할 수 있습니다.

./bin/generate-example-metrics | python /quickstart/protobuf/pb_publisher.py

Kafka 설치 디렉토리에서 다음 명령으로 데이터가 Kafka 토픽에 삽입됐는지 확인할 수 있어요.

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic metrics_pb --from-beginning

그러면 다음과 같은 메시지가 출력됩니다.

millisecondsGETR"2017-04-06T03:23:56Z*2002/list:request/latencyBwww1.example.com

이전 단계에서 만든 supervisor가 실행 중이라면 indexing task가 메시지를 생산하기 시작하고, 곧 Druid에서 데이터를 쿼리할 수 있게 됩니다.

예시 파일 생성하기

예시 quickstart에 제공된 파일들은 metrics.proto만으로 시작해서 다음과 같은 방식으로 만들 수 있어요.

metrics.desc

descriptor 파일은 protoc Protobuf 컴파일러로 생성됩니다. .proto 파일이 주어지면 다음과 같이 .desc 파일을 만들 수 있어요.

protoc -o metrics.desc metrics.proto

metrics_pb2.py

metrics_pb2.py도 protoc로 생성합니다.

 protoc -o metrics.desc metrics.proto --python_out=.

pb_publisher.py

metrics_pb2.py가 생성된 후, JSON 데이터를 파싱해서 Protobuf로 변환하고 Kafka 토픽으로 produce하는 또 다른 스크립트를 만들 수 있어요.

#!/usr/bin/env pythonimport sysimport jsonfrom kafka import KafkaProducerfrom metrics_pb2 import Metricsproducer = KafkaProducer(bootstrap_servers='localhost:9092')topic = 'metrics_pb'for row in iter(sys.stdin):    d = json.loads(row)    metrics = Metrics()    for k, v in d.items():        setattr(metrics, k, v)    pb = metrics.SerializeToString()    producer.send(topic, pb)producer.flush()

더 알아보기 (Learn more)