外部データソース¶

DataFlux Func を介して、MySQL、Prometheus など、さまざまなタイプの外部データソースを Guance に接続できます。また、TiDB Cloud Lake への直接接続もサポートしており、データの統合クエリと可視化を実現します。

機能¶

ネイティブクエリ：グラフ内でデータソースのネイティブクエリステートメントを直接使用できるため、追加の変換は一切不要です。
データ保護：Func データソースの接続情報はローカルの Func に保存されます。TiDB Cloud Lake の Lake DSN はプラットフォームによって暗号化されて保存され、パスワードと完全な DSN はリスト、詳細、クエリで表示されることはありません。
カスタム管理：実際のニーズに応じて、さまざまな外部データソースを簡単に追加および管理できます。
リアルタイムデータ：外部データソースに直接接続してリアルタイムにデータを取得し、即座に応答して意思決定を行うことができます。

接続方法¶

接続方法	対応データソース	説明
DataFlux Func を介して接続	MySQL、Prometheus など	既にデプロイされている DataFlux Func と Connector を再利用します。ID などの既存の設定は変更しないでください。
TiDB Cloud Lake 直接接続	TiDB Cloud Lake	Guance サーバーが TiDB Cloud Lake Driver を介して接続します。DataFlux Func に依存せず、データソース ID の入力も不要です。

TiDB Cloud Lake に直接接続する必要がある場合は、TiDB Cloud Lake への接続とクエリを参照してください。

DataFlux Func を介して接続する¶

Guance 側でデータソースを追加する¶

拡張で直接、接続された DataFlux Func を追加または表示し、接続されているすべての外部データソースをさらに管理します。

注意

この方法は、2 番目のパスと比較して初心者にも優しいため、推奨されます。

プルダウンで DataFlux Func を選択します。
サポートされているデータソースのタイプを選択します。
接続プロパティ（ID、データソースのタイトル、関連付けるホスト、ポート、データベース、ユーザー、パスワード）を定義します。
必要に応じて接続をテストします。
保存します。

Func を使用して外部データソースをクエリする¶

注意

ここで言う「外部データソース」は広義の定義であり、一般的な外部データ保存システム（MySQL、Redis などのデータベース）だけでなく、サードパーティシステム（Guance コンソールなど）も含まれます。

前提条件

対応するインストールパッケージをダウンロードし、クイックスタートに従って Func プラットフォームをデプロイする必要があります。

デプロイ完了後、初期化が完了するのを待って、プラットフォームにログインします。

Func と Guance を関連付ける¶

コネクタを使用すると、開発者は Guance システムに接続できます。

開発 > コネクタ > コネクタの追加ページに移動します：

コネクタのタイプを選択します。
このコネクタの ID をカスタマイズします。
タイトルを追加します。このタイトルは Guance ワークスペース内に同期して表示されます。
必要に応じて、このコネクタの説明を入力します。
Guance ノードを選択します。
API Key ID と API Key を追加します。
必要に応じて接続性をテストします。
保存します。

関連付けが完了すると、Func プラットフォームでは次の 2 つの方法でデータソースをクエリできます。

コネクタを使用する
スクリプトを自作する

API Key の取得方法¶

Guance ワークスペース > 管理 > API Key 管理に移動します。
ページ右側の Key を新規作成 をクリックします。
名前を入力します。
確定をクリックします。システムが自動的に API Key を作成します。API Key 内で確認できます。

詳細については、API Key 管理を参照してください。

コネクタを使用する¶

コネクタを正常に追加すると、スクリプト内でコネクタ ID を使用して、対応するコネクタの操作オブジェクトを取得できます。

上記のコネクタを例に、このコネクタの操作オブジェクトを取得するコードは次のとおりです：

mysql = DFF.CONN('mysql')

スクリプトを自作する¶

コネクタを使用する以外に、関数を自作してデータをクエリすることもできます。

ユーザーが MySQL コネクタ（ID を mysql と定義）を正しく作成し、この MySQL に my_table という名前のテーブルが存在し、以下のデータが含まれていると仮定します：

`id`	`userId`	`username`	`reqMethod`	`reqRoute`	`reqCost`	`createTime`
1	u-001	admin	POST	/api/v1/scripts/:id/do/modify	23	1730840906
2	u-002	admin	POST	/api/v1/scripts/:id/do/publish	99	1730840906
3	u-003	zhang3	POST	/api/v1/scripts/:id/do/publish	3941	1730863223
4	u-004	zhang3	POST	/api/v1/scripts/:id/do/publish	159	1730863244
5	u-005	li4	POST	/api/v1/scripts/:id/do/publish	44	1730863335
...

ここで、データクエリ関数 を使用してこのテーブルのデータをクエリし、フィールド抽出ルールが次のとおりであると仮定します：

元のフィールド	抽出先
`createTime`	時間 `time`
`reqCost`	列 `req_cost`
`reqMethod`	列 `req_method`
`reqRoute`	列 `req_route`
`userId`	タグ `user_id`
`username`	タグ `username`

この場合、完全な参考コードは次のとおりです：

データクエリ関数の例

import json

@DFF.API('my_table からのデータをクエリ', category='dataPlatform.dataQueryFunc')
def query_from_my_table(time_range):
    # コネクタ操作オブジェクトを取得
    mysql = DFF.CONN('mysql')

    # MySQL クエリステートメント
    sql = '''
      SELECT
        createTime, userId, username, reqMethod, reqRoute, reqCost
      FROM
        my_table
      WHERE
        createTime     > ?
        AND createTime < ?
      LIMIT 5
    '''

    # 渡された time_range の単位はミリ秒
    # しかし MySQL の createTime フィールドの単位は秒であるため、変換が必要
    sql_params = [
      int(time_range[0] / 1000),
      int(time_range[1] / 1000),
    ]

    # クエリを実行
    db_res = mysql.query(sql, sql_params)

    # DQL ライクな戻り値に変換

    # タグが異なる場合、複数のデータ系列を生成する必要がある場合がある
    # データ系列タグを Key としてマッピングテーブルを作成
    series_map = {}

    # 元のデータを反復処理し、構造を変換してマッピングテーブルに格納
    for d in db_res:
        # タグを収集
        tags = {
            'user_id' : d.get('userId'),
            'username': d.get('username'),
        }

        # タグをシリアライズ（tag key をソートして出力を一貫させる）
        tags_dump = json.dumps(tags, sort_keys=True, ensure_ascii=True)

        # このタグのデータ系列がまだ作成されていない場合は作成する
        if tags_dump not in series_map:
            # データ系列 series の基本構造
            series_map[tags_dump] = {
                'columns': [ 'time', 'req_cost', 'req_method', 'req_route' ], # 列（1 列目は固定で time）
                'tags'   : tags,                                              # タグ
                'values' : [],                                                # 値リスト
            }

        # 時間、列を抽出し、value を追加
        series = series_map[tags_dump]
        value = [
            d.get('createTime') * 1000, # 時間 time（出力単位はミリ秒、実際に応じて変換が必要）
            d.get('reqCost'),           # 列 req_cost
            d.get('reqMethod'),         # 列 req_method
            d.get('reqRoute'),          # 列 req_route
        ]
        series['values'].append(value)

    # DQL の外側の構造を追加
    dql_like_res = {
        # データ系列
        'series': [ list(series_map.values()) ] # ここでさらに 1 段階配列でラップすることに注意
    }
    return dql_like_res

データ変換処理のみを理解したい場合や、クエリプロセスを気にしない場合（または実際にクエリ可能なデータベースがまだない場合）は、次のコードを参照してください：

データクエリ関数の例（MySQL クエリ部分を含まない）

import json

@DFF.API('某所からのデータをクエリ', category='dataPlatform.dataQueryFunc')
def query_from_somewhere(time_range):
    # 何らかの方法で元のデータを取得したと仮定
    db_res = [
        {'createTime': 1730840906, 'reqCost': 23,   'reqMethod': 'POST', 'reqRoute': '/api/v1/scripts/:id/do/modify',  'username': 'admin',  'userId': 'u-001'},
        {'createTime': 1730840906, 'reqCost': 99,   'reqMethod': 'POST', 'reqRoute': '/api/v1/scripts/:id/do/publish', 'username': 'admin',  'userId': 'u-001'},
        {'createTime': 1730863223, 'reqCost': 3941, 'reqMethod': 'POST', 'reqRoute': '/api/v1/scripts/:id/do/publish', 'username': 'zhang3', 'userId': 'u-002'},
        {'createTime': 1730863244, 'reqCost': 159,  'reqMethod': 'POST', 'reqRoute': '/api/v1/scripts/:id/do/publish', 'username': 'zhang3', 'userId': 'u-002'},
        {'createTime': 1730863335, 'reqCost': 44,   'reqMethod': 'POST', 'reqRoute': '/api/v1/scripts/:id/do/publish', 'username': 'li4',    'userId': 'u-003'}
    ]

    # DQL ライクな戻り値に変換

    # タグが異なる場合、複数のデータ系列を生成する必要がある場合がある
    # データ系列タグを Key としてマッピングテーブルを作成
    series_map = {}

    # 元のデータを反復処理し、構造を変換してマッピングテーブルに格納
    for d in db_res:
        # タグを収集
        tags = {
            'user_id' : d.get('userId'),
            'username': d.get('username'),
        }

        # タグをシリアライズ（tag key をソートして出力を一貫させる）
        tags_dump = json.dumps(tags, sort_keys=True, ensure_ascii=True)

        # このタグのデータ系列がまだ作成されていない場合は作成する
        if tags_dump not in series_map:
            # データ系列 series の基本構造
            series_map[tags_dump] = {
                'columns': [ 'time', 'req_cost', 'req_method', 'req_route' ], # 列（1 列目は固定で time）
                'tags'   : tags,                                              # タグ
                'values' : [],                                                # 値リスト
            }

        # 時間、列を抽出し、value を追加
        series = series_map[tags_dump]
        value = [
            d.get('createTime') * 1000, # 時間 time（出力単位はミリ秒、実際に応じて変換が必要）
            d.get('reqCost'),           # 列 req_cost
            d.get('reqMethod'),         # 列 req_method
            d.get('reqRoute'),          # 列 req_route
        ]
        series['values'].append(value)

    # DQL の外側の構造を追加
    dql_like_res = {
        # データ系列
        'series': [ list(series_map.values()) ] # ここでさらに 1 段階配列でラップすることに注意
    }
    return dql_like_res

戻り値の例

{
  "series": [
    [
      {
        "columns": ["time", "req_cost", "req_method", "req_route"],
        "tags": {"user_id": "u-001", "username": "admin"},
        "values": [
          [1730840906000, 23, "POST", "/api/v1/scripts/:id/do/modify" ],
          [1730840906000, 99, "POST", "/api/v1/scripts/:id/do/publish"]
        ]
      },
      {
        "columns": ["time", "req_cost", "req_method", "req_route"],
        "tags": {"user_id": "u-002", "username": "zhang3"},
        "values": [
          [1730863223000, 3941, "POST", "/api/v1/scripts/:id/do/publish"],
          [1730863244000,  159, "POST", "/api/v1/scripts/:id/do/publish"]
        ]
      },
      {
        "columns": ["time", "req_cost", "req_method", "req_route"],
        "tags": {"user_id": "u-003", "username": "li4"},
        "values": [
          [1730863335000, 44, "POST", "/api/v1/scripts/:id/do/publish"]
        ]
      }
    ]
  ]
}

管理リスト¶

接続されているすべてのデータソースは、インテグレーション > 外部データソース > 接続済みデータソース で確認できます。

リストでは、次の操作を実行できます。

データソースのタイプ、ID、ステータス、作成情報、更新情報などを表示します。
特定のデータソースを編集します。DataFlux Func、データソースのタイプ、ID 以外の設定を変更できます。
特定のデータソースを削除します。

ユースケース¶

Guance で外部データソースを使用してクエリを実行する典型的なシナリオの 1 つは、グラフ > グラフクエリです。

異なるグラフのデータ戻り値
折れ線グラフ	円グラフ	テーブルグラフ