integrations/sagemaker/timestreamquery.py [30:105]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        client = session.client(service_name = 'timestream-query',
                                region_name = region, config = config)

    return client

def parseDatum(c_type, data):
    if ('ScalarType' in c_type):
        return parseScalar(c_type['ScalarType'], data.get('ScalarValue'))
    elif ('ArrayColumnInfo' in c_type):
        return parseArrayData(c_type['ArrayColumnInfo'], data.get('ArrayValue'))
    elif ('TimeSeriesMeasureValueColumnInfo' in c_type):
        return parseTSData(c_type['TimeSeriesMeasureValueColumnInfo'], data.get('TimeSeriesValue'))
    elif ('RowColumnInfo' in c_type):
        return parseRowData(c_type['RowColumnInfo'], data.get('RowValue'))
    else:
        raise Exception("All the data is Null???")

def parseScalar(c_type, data):
    if data == None:
        return None
    if (c_type == "VARCHAR"):
        return data
    elif (c_type == "BIGINT"):
        return int(data)
    elif (c_type == "DOUBLE"):
        return float(data)
    elif (c_type == "INTEGER"):
        return int(data)
    elif (c_type == "BOOLEAN"):
        return bool(data)
    elif (c_type == "TIMESTAMP"):
        return data
    else:
        return data

def parseArrayData(c_type, data):
    if data == None:
        return None
    datum_list = []
    for elem in data:
        datum_list.append(parseDatum(c_type['Type'], elem))
    return datum_list

def parseTSData(c_type, data):
    if data == None:
        return None
    datum_list = []
    for elem in data:
        ts_data = {}
        ts_data['time'] = elem['Time']
        ts_data['value'] = parseDatum(c_type['Type'], elem['Value'])
        datum_list.append(ts_data)
    return datum_list

def parseRowData(c_types, data):
    if data == None:
        return None
    datum_dict = {}
    for c_type, elem in zip(c_types, data['Data']):
        datum_dict[c_type['Name']] = parseDatum(c_type['Type'], elem)
    return datum_dict

def flatModelToDataframe(items):
    """
    Translate a Timestream query SDK result into a Pandas dataframe.
    """
    return_val = defaultdict(list)
    for obj in items:
        for row in obj.get('Rows'):
            for c_info, data in zip(obj['ColumnInfo'], row['Data']):
                c_name = c_info['Name']
                c_type = c_info['Type']
                return_val[c_name].append(parseDatum(c_type, data))

    df = pd.DataFrame(return_val)
    return df
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



tools/perf-scale-workload/timestreamquery.py [38:119]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
            client = session.client(service_name = 'timestream-query',
                region_name = region, config = config)

    return client

##################################################
#### Methods to parse the query results ##########
##################################################

def parseDatum(c_type, data):
    if ('ScalarType' in c_type):
        return parseScalar(c_type['ScalarType'], data.get('ScalarValue'))
    elif ('ArrayColumnInfo' in c_type):
        return parseArrayData(c_type['ArrayColumnInfo'], data.get('ArrayValue'))
    elif ('TimeSeriesMeasureValueColumnInfo' in c_type):
        return parseTSData(c_type['TimeSeriesMeasureValueColumnInfo'], data.get('TimeSeriesValue'))
    elif ('RowColumnInfo' in c_type):
        return parseRowData(c_type['RowColumnInfo'], data.get('RowValue'))
    else:
        raise Exception("All the data is Null???")

def parseScalar(c_type, data):
    if data == None:
        return None
    if (c_type == "VARCHAR"):
        return data
    elif (c_type == "BIGINT"):
        return int(data)
    elif (c_type == "DOUBLE"):
        return float(data)
    elif (c_type == "INTEGER"):
        return int(data)
    elif (c_type == "BOOLEAN"):
        return bool(data)
    elif (c_type == "TIMESTAMP"):
        return data
    else:
        return data

def parseArrayData(c_type, data):
    if data == None:
        return None
    datum_list = []
    for elem in data:
        datum_list.append(parseDatum(c_type['Type'], elem))
    return datum_list

def parseTSData(c_type, data):
    if data == None:
        return None
    datum_list = []
    for elem in data:
        ts_data = {}
        ts_data['time'] = elem['Time']
        ts_data['value'] = parseDatum(c_type['Type'], elem['Value'])
        datum_list.append(ts_data)
    return datum_list

def parseRowData(c_types, data):
    if data == None:
        return None
    datum_dict = {}
    for c_type, elem in zip(c_types, data['Data']):
        datum_dict[c_type['Name']] = parseDatum(c_type['Type'], elem)
    return datum_dict

## Convert the flat results to a pandas dataframe.
def flatModelToDataframe(items):
    """
    Translate a Timestream query SDK result into a Pandas dataframe.
    """
    return_val = defaultdict(list)
    for obj in items:
        for row in obj.get('Rows'):
            for c_info, data in zip(obj['ColumnInfo'], row['Data']):
                c_name = c_info['Name']
                c_type = c_info['Type']
                return_val[c_name].append(parseDatum(c_type, data))

    #print(json.dumps(return_val, indent=4))
    df = pd.DataFrame(return_val)
    return df
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -