이벤트 데이터 샤딩 실전 가이드: Dataway Sink 기반 구현方案¶

본 문서는 DataFlux Func를 통한 HTTP Header 주입 및 Dataway Sinker 규칙 구성을 통해 이벤트 데이터(keyevent)를 지능적으로 샤딩하는 방법을 상세히 설명합니다. 이方案을 통해 다양한 비즈니스 속성 및 환경 특성을 가진 이벤트 데이터를 지정된 워크스페이스로 라우팅할 수 있습니다.

方案 원리¶

데이터 샤딩 흐름¶

핵심 메커니즘 설명¶

DataFlux Func 측 식별자 주입: 이벤트 데이터가 보고될 때 Func 구성을 통해 X-Global-Tags Header를 동적으로 생성하며, 샤딩에 필요한 키-값 쌍(예: env=prod)을 포함합니다.
Dataway 라우팅 매칭: Dataway는 sinker.json에 정의된 규칙에 따라 특정 식별자를 가진 이벤트를 해당 워크스페이스로 전달합니다.

1. Dataway 구성¶

이 기능을 사용하기 전에 Dataway가 배포되어 있고 Sinker 샤딩 기능이 활성화되어 있는지 확인하십시오.

Sinker 구성에 대한 자세한 내용은 Dataway Sinker 구성 가이드를 참조하십시오.

참고: 배포 버전에 내장된 DataFlux Func에서 사용하는 Dataway는 utils 네임스페이스 아래의 internal-dataway에 있습니다.

2. DataFlux Func 구성¶

Header 주입 X-Global-Tags¶

핵심 매개변수 설명¶

매개변수명	유형	설명
`CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS`	list/string	이벤트 데이터의 샤딩 식별자 생성 규칙 정의

간단한 예제¶

모든 워크스페이스의 이벤트를 "이벤트 중앙 관리" 워크스페이스에 통합하여 기록:

Launcher 콘솔에 접속합니다.
우측 상단 > 애플리케이션 구성 수정으로 이동합니다.
func2 네임스페이스 아래의 func2Config 구성 항목을 찾습니다.

구성을 추가합니다:

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: keyevent     # 데이터 카테고리
    fields: df_source      # 샤딩에 사용할 필드, 여기에는 이벤트의 고정 식별 필드를 입력

Dataway Sinker 규칙 구성: sinker.json 구성 파일을 수정하여 데이터 라우팅 규칙을 설정합니다:

{
    "strict": true,
    "rules": [
        {
            "rules": [
                "{ df_source = 'monitor' }"
            ],
            "url": "워크스페이스 데이터 보고 주소"
        }
    ]
}

특수 필드 설명¶

필드명	설명
`DF_WORKSPACE_UUID`	워크스페이스 ID
`DF_WORKSPACE_NAME`	워크스페이스 이름
`DF_MONITOR_CHECKER_ID`	모니터 ID
`DF_MONITOR_CHECKER_NAME`	모니터 이름

추가 고급 구성¶

구성 방식	예제	설명
직접 추출	`-host`	이벤트 데이터의 `tags` 또는 `fields`에서 `host` 필드 추출
필드 이름 변경	`-src:service; dest:business_type`	`service` 필드를 `business_type`으로 이름 변경
값 매핑	`remap:{order:电商业务}`	원래 값 `order`를 `电商业务`로 매핑
기본값	`default:unknown`	필드가 없을 때 기본값 사용
고정값	`- dest:env; fixed:prod`	고정값 `env=prod` 직접 주입

Global Tags 생성 규칙¶

필드명	유형	기본값	설명
`[#].category`	string/[string]	`"*"`	매칭할 데이터 카테고리
`[#].fields`	string/dict [string]/[dict]	-	데이터 필드(Tags 및 Fields 포함) 추출; 직접 추출 및 규칙 기반 추출 지원
`[#].fields[#]`	string	-	추출할 필드명, 추가 추출 필드 지원(아래 표 참조)
`[#].fields[#]`	dict	-	추출 필드 규칙
`[#].fields[#].src`	string	-	추출할 필드명, 추가 추출 필드 지원(아래 표 참조)
`[#].fields[#].dest`	string	`src`와 동일	추출 후 Header에 기록할 필드명
`[#].fields[#].default`	string	-	지정된 필드가 없을 때 Header에 기록할 기본값
`[#].fields[#].fixed`	string	-	Header에 기록할 고정값
`[#].fields[#].remap`	dict	`null`	추출된 필드 값에 대한 매핑 변환
`[#].fields[#].remap_default`	string	-	추출된 필드 값 매핑 변환 시 해당 매핑 값이 없을 때의 기본값 지정하지 않으면 원래 값 유지 `null`로 지정하면 이 필드 무시
`[#].filter`	dict/string	`null`	데이터 매칭 필터 Tag 필터 및 filterString 필터 지원

사용자 정의 Global Tags 생성 함수 ID¶

함수 ID 형식은 {스크립트 세트 ID}__{스크립트 ID}.{함수명}입니다.

함수 정의는 다음과 같습니다:

매개변수	유형	설명
`category`	string	카테고리, 예: `"keyevent"`
`point`	dict	처리할 단일 데이터
`point.measurement`	string	데이터 `measurement`
`point.tags`	dict	데이터 `tags` 내용
`point.fields`	dict	데이터 `fields` 내용
`extra_fields`	dict	추가 추출 필드(아래 표 참조)

예제:

point 매개변수 값

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001",
    "ip"  : "1.2.3.4"
  },
  "fields": {
    "name": "Tom"
  }
}

extra_fields 매개변수 값

{
  "DF_WORKSPACE_UUID"      : "wksp_xxxxx",
  "DF_MONITOR_CHECKER_ID"  : "rul_xxxxx",
  "DF_MONITOR_CHECKER_NAME": "모니터 XXXXX",
  "DF_WORKSPACE_NAME"      : "워크스페이스 XXXXX"
}

생성 효과 검증¶

Header에 key:value를 추가하는 방식 예제 {#example}

이벤트 데이터를 동일한 워크스페이스에 기록¶

이벤트에서 필드 추출

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: keyevent
    fields:
      - host
      - name
      - DF_WORKSPACE_UUID

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001",
    "ip"  : "1.2.3.4"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: host=web-001,name=Tom,DF_WORKSPACE_UUID=wksp_xxxxx

이벤트에서 단일 필드 추출

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: keyevent
    # 필드가 1개인 경우 간략히 작성 가능
    fields: host

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: host=web-001

모든 데이터를 동일한 워크스페이스에 기록¶

category를 작성하지 않으면 모든 데이터를 처리합니다.

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields: DF_WORKSPACE_UUID

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: DF_WORKSPACE_UUID=wksp_xxxxx

기타 상황¶

필드 추출 시 필드명 변경

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields:
    - src : host
      dest: HOST

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: HOST=web-001

필드 추출 시 필드 값 매핑

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields:
    - src : result
      remap:
        OK     : ok
        success: ok
        failed : error
        failure: error
        timeout: error
      remap_default: unknown

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "result": "success"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: result=ok

필드 추출 시 기본값 사용

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields:
    - src    : result
      default: unknown

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: result=unknown

고정값 기록

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields:
    - dist : app
      fixed: dataPlatform

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "host": "web-001"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: app=dataPlatform

Tag 방식으로 데이터 매칭

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields: host
    filter:
      service: app-*
  - fields: client_ip
    filter:
      service: web-*

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "host"     : "app-001",
    "client_ip": "1.2.3.4",
    "service"  : "app-user"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: host=app-001

filterString 방식으로 데이터 매칭

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - fields: host
    filter: 'service:app-*'
  - fields: client_ip
    filter: 'service:web-*'

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "host"     : "app-001",
    "client_ip": "1.2.3.4",
    "service"  : "app-user"
  },
  "fields": {
    "name": "Tom"
  }
}

예제로 기록된 Header

X-Global-Tags: host=app-001

사용자 정의 함수 방식으로 이벤트 필드 접두사/접미사 추출

예제 구성

CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS: my_script_set__my_script.make_global_tags

예제 함수 (스크립트 세트 my_script_set, 스크립트 my_script 아래)

def make_global_tags(category, point, extra_fields):
    # 이벤트 유형 데이터만 처리
    if category != 'keyevent':
        return

    global_tags_list = {}

    # 데이터의 fields 또는 tags에서 name, region 필드 가져오기
    name   = point['fields'].get('name')   or point['tags'].get('name')
    region = point['fields'].get('region') or point['tags'].get('region')

    # name 접두사 가져오기
    if name:
        prefix = str(name).split('-')[0]
        global_tags_list['name_prefix'] = prefix

    # region 접미사 가져오기
    if region:
        suffix = str(region).split('-').pop()
        global_tags_list['region_suffix'] = suffix

    # 반환
    return global_tags_list

예제 데이터

{
  "measurement": "keyevent",
  "tags": {
    "region"   : "cn-shanghai",
    "service"  : "app-user"
  },
  "fields": {
    "name": "Tom-Jerry"
  }
}

예제로 기록된 Header

X-Global-Tags: name_prefix=Tom,region_suffix=shanghai

이벤트 보고 예제:

{
  "measurement": "keyevent",
  "tags": { "host": "web-01", "service": "order" },
  "fields": { "message": "사용자 주문 오류" }
}

생성된 HTTP Header:

X-Global-Tags: host=web-01,business_type=电商业务,DF_WORKSPACE_UUID=wksp_123

3. Dataway Sinker 규칙 구성¶

규칙 파일 예제 (`sinker.json`)¶

{
  "strict": false,
  "rules": [
    {
      "rules": ["{ business_type = '电商业务' }"],  // 전자상거래 비즈니스 이벤트 매칭
      "url": "https://kodo.guance.com?token=tkn_电商空间令牌"
    },
    {
      "rules": ["{ DF_WORKSPACE_UUID = 'wksp_123' }"],  // 지정된 워크스페이스 매칭
      "url": "https://backup.guance.com?token=tkn_备份空间令牌"
    },
    {
      "rules": ["*"],  // 기본 규칙 (반드시 존재해야 함)
      "url": "https://default.guance.com?token=tkn_默认空间令牌"
    }
  ]
}

규칙 구문 설명¶

자세한 연산자는 필터 연산자를 참조하십시오.

4. Datakit 측 구성 설명¶

기본 구성¶

# /usr/local/datakit/conf.d/datakit.conf
[dataway]
  # Sinker 기능 활성화
  enable_sinker = true
  # 샤딩 기준 필드 정의 (최대 3개)
  global_customer_keys = ["host", "env"]

주의사항¶

필드 유형 제한: 문자열 유형 필드만 지원합니다 (모든 Tag 값은 문자열입니다).
바이너리 데이터 지원: 세션 리플레이, 프로파일링 등 바이너리 데이터 샤딩을 지원합니다.
성능 영향: 샤딩 필드가 하나 추가될 때마다 메모리 사용량이 약 5% 증가합니다.

5. 글로벌 Tag의 영향¶

1. 글로벌 Tag 예제¶

# datakit.conf
[election.tags]
    cluster = "cluster-A"  # 글로벌 선출 Tag
[global_tags]
    region = "cn-east"     # 글로벌 호스트 Tag

2. 샤딩 식별자 병합 로직¶

이벤트 데이터에 다음 Tag가 포함되어 있다고 가정합니다:

{
  "tags": { "cluster": "cluster-B", "app": "payment" }
}

최종 샤딩 식별자:

X-Global-Tags: cluster=cluster-B,region=cn-east

확장 설명: 기타 데이터 유형 샤딩¶

1. 사용자 정의 샤딩 규칙¶

이벤트 데이터가 아닌 데이터(예: logging, metric)의 경우 category를 지정하여 샤딩을 구현합니다:

# Func 구성 예제: logging 데이터 처리
CUSTOM_INTERNAL_DATAWAY_X_GLOBAL_TAGS:
  - category: logging
    fields:
      - src: log_level
        remap:
          error: 중요 오류
          warn: 일반 경고
      - service

2. 일반 원칙¶

구성 격리: 서로 다른 데이터 카테고리(keyevent/logging/metric)는 독립적인 구성 블록을 사용합니다.
필드 간소화: 단일 데이터 카테고리의 샤딩 식별자는 3개를 초과하지 않습니다.
충돌 방지: 서로 다른 카테고리의 샤딩 필드는 서로 다른 이름을 사용하는 것이 좋습니다.

문제 해결¶

일반적인 문제¶

현상	문제 해결 단계
샤딩이 적용되지 않음	1. Dataway 로그 확인 `grep 'sinker reload'` 2. `curl -v`로 Header 검증 3. Sinker 규칙 우선순위 확인
일부 데이터 손실	1. `strict` 모드 상태 확인 2. 기본 규칙 존재 여부 확인
식별자가 주입되지 않음	1. Func 구성 구문 검증 2. 필드가 문자열 유형인지 확인

진단 명령어¶

# Dataway 샤딩 통계 확인
curl http://localhost:9528/metrics | grep sinker_requests_total
# 샤딩 규칙 수동 테스트
curl -X POST -H "X-Global-Tags: business_type=电商业务" http://dataway/v1/write/keyevent

이벤트 데이터 샤딩 실전 가이드: Dataway Sink 기반 구현方案¶

方案 원리¶

데이터 샤딩 흐름¶

핵심 메커니즘 설명¶

1. Dataway 구성¶

2. DataFlux Func 구성¶

Header 주입 X-Global-Tags¶

핵심 매개변수 설명¶

간단한 예제¶

특수 필드 설명¶

추가 고급 구성¶

Global Tags 생성 규칙¶

사용자 정의 Global Tags 생성 함수 ID¶

생성 효과 검증¶

이벤트 데이터를 동일한 워크스페이스에 기록¶

모든 데이터를 동일한 워크스페이스에 기록¶

기타 상황¶

3. Dataway Sinker 규칙 구성¶

규칙 파일 예제 (sinker.json)¶

규칙 구문 설명¶

4. Datakit 측 구성 설명¶

기본 구성¶

주의사항¶

5. 글로벌 Tag의 영향¶

1. 글로벌 Tag 예제¶

2. 샤딩 식별자 병합 로직¶

확장 설명: 기타 데이터 유형 샤딩¶

1. 사용자 정의 샤딩 규칙¶

2. 일반 원칙¶

문제 해결¶

일반적인 문제¶

진단 명령어¶

이 페이지가 도움이 되었나요? ×

규칙 파일 예제 (`sinker.json`)¶

이 페이지가 도움이 되었나요?