イベントデータシャーディング実践：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 値は文字列です）
バイナリデータのサポート：Session Replay、Profiling などのバイナリデータのシャーディングをサポートします
パフォーマンスへの影響：シャーディングフィールドが1つ増えるごとに、メモリ使用量が約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`)¶

このページは役に立ちましたか？