외부 데이터 소스¶

mysql = DFF.CONN('mysql')

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()) ] # 여기에 배열을 한 번 더 감싸야 함
    }
    return dql_like_res

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()) ] # 여기에 배열을 한 번 더 감싸야 함
    }
    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"]
        ]
      }
    ]
  ]
}