外部データソース¶

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'),
        }

        # タグをシリアル化（タグキーをソートして出力の一貫性を保証）
        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()) ] # ここではさらに配列でラップすることに注意
    }
    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'),
        }

        # タグをシリアル化（タグキーをソートして出力の一貫性を保証）
        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()) ] # ここではさらに配列でラップすることに注意
    }
    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 つは、グラフ > グラフクエリです。

異なるグラフのデータ返却
折れ線グラフ	円グラフ	テーブルグラフ