AWS Timestreamからデータを取得するLambda関数のエラー

Question

前提

AWS IoTの以下の公式ブログを参考に、デジタルツインを構築し、センサデータの可視化を目指しております。

”IoT デバイスのデジタルツインを構築し、AWS IoT TwinMaker を使用してリアルタイムのセンサーデータを監視する (パート 2/2)”
https://aws.amazon.com/jp/blogs/news/build-a-digital-twin-of-your-iot-device-and-monitor-real-time-sensor-data-using-aws-iot-twinmaker-part-2-of-2/

Timestreamに保存されたデータをLambda関数を用いてIoT Twinmakerで参照しようとしたら、エラーが発生しました。

実現したいこと

エラーが発生しているLambda関数を正常に動かし、
ブログの通りTwinmakerおよびGrafana にてセンサデータの可視化を行いたい。

発生している問題・エラーメッセージ

ブログのpart1，2同様の環境を構築し、part2の5．最終結果まで手順を進め、デバイス側でPythonスクリプトを実行してダッシュボードで値を可視化しようとしたら、ダッシュボードの左上に”!”マークが出て、以下のエラーが表示され値が反映されていないです。

ConnectorFailureException: Lambda function arn:aws:lambda:[region]:[account id]:function:timestreamReader throws exception. Please check your lambda logs for detailed information.

[region]と[account id]には私のAWS情報が記載されております。

Lambda関数がうまく動いていないようなので、CloudWatchにて該当箇所のエラーの中身も確認しました。

[ERROR] Runtime.UserCodeSyntaxError: Syntax error in module 'lambda_function': EOL while scanning string literal (lambda_function.py, line 122)
Traceback (most recent call last):
  File "/var/task/lambda_function.py" Line 122
        entity_property_reference_temp['propertyName'] = 'temperature’

該当のソースコード

以下、Lambda関数の中身です。

import logging
import json
import os
import boto3

from datetime import datetime

LOGGER = logging.getLogger()
LOGGER.setLevel(logging.INFO)

# Get db and table name from Env variables
DATABASE_NAME = os.environ['TIMESTREAM_DATABASE_NAME']
TABLE_NAME = os.environ['TIMESTREAM_TABLE_NAME']

# Python boto client for AWS Timestream
QUERY_CLIENT = boto3.client('timestream-query')


# Utility function: parses a timestream row into a python dict for more convenient field access
def parse_row(column_schema, timestream_row):
    data = timestream_row['Data']
    result = {}
    for i in range(len(data)):
        info = column_schema[i]
        datum = data[i]
        key, val = parse_datum(info, datum)
        result[key] = val
    return result

# Utility function: parses timestream datum entries into (key,value) tuples. Only ScalarTypes currently supported.
def parse_datum(info, datum):
    if datum.get('NullValue', False):
        return info['Name'], None
    column_type = info['Type']
    if 'ScalarType' in column_type:
        return info['Name'], datum['ScalarValue']
    else:
        raise Exception(f"Unsupported columnType[{column_type}]")

# This function extracts the timestamp from a Timestream row and returns in ISO8601 basic format
def get_iso8601_timestamp(str):
    #  e.g. '2022-04-06 00:17:45.419000000' -> '2022-04-06T00:17:45.419000000Z'
    return str.replace(' ', 'T') + 'Z'

# Main logic
def lambda_handler(event, context):
    selected_property = event['selectedProperties'][0]

    LOGGER.info("Selected property is %s", selected_property)

    # 1. EXECUTE THE QUERY TO RETURN VALUES FROM DATABASE
    query_string = f"SELECT measure_name, time, measure_value::bigint" \
        f" FROM {DATABASE_NAME}.{TABLE_NAME} " \
        f" WHERE time > from_iso8601_timestamp('{event['startTime']}')" \
        f" AND time <= from_iso8601_timestamp('{event['endTime']}')" \
        f" AND measure_name = '{selected_property}'" \
        f" ORDER BY time ASC"
            
    try:
        query_page = QUERY_CLIENT.query(
            QueryString = query_string
        )
    except Exception as err:
        LOGGER.error("Exception while running query: %s", err)
        raise err

    # Query result structure: https://docs.aws.amazon.com/timestream/latest/developerguide/API_query_Query.html

    next_token = None
    if query_page.get('NextToken') is not None:
       next_token = query_page['NextToken']
    schema = query_page['ColumnInfo']

    # 2. PARSE TIMESTREAM ROWS
    result_rows = []
    for row in query_page['Rows']:
        row_parsed = parse_row(schema,row)
        #LOGGER.info('row parsed: %s', row_parsed)
        result_rows.append(row_parsed)

    # 3. CONVERT THE QUERY RESULTS TO THE FORMAT TWINMAKER EXPECTS

    # There must be one entityPropertyReference for Humidity OR one for Temperature
    entity_property_reference_temp = {}
    entity_property_reference_temp['componentName'] = 'timestream-reader'
    entity_property_reference_temp['propertyName'] = 'temperature’
    entity_property_reference_temp['entityId'] = 'xxx'


    entity_property_reference_hum = {}
    entity_property_reference_hum['componentName'] = 'timestream-reader'
    entity_property_reference_hum['propertyName'] = 'humidity’
    entity_property_reference_hum['entityId'] = 'xxx'


    values_temp = []
    values_hum = []

    for result_row in result_rows:
        ts = result_row['time']
        measure_name = result_row['measure_name']
        measure_value = result_row['measure_value::bigint']

        time = get_iso8601_timestamp(ts)
        value = { 'doubleValue' : str(measure_value) }

        if measure_name == 'temperature':
            values_temp.append({
                'time': time,
                'value':  value
            })
        elif measure_name == 'humidity':
             values_hum.append({
                'time': time,
                'value':  value
            })

    # The final structure "propertyValues"
    property_values = []

    if(measure_name == 'temperature'):
        property_values.append({
            'entityPropertyReference': entity_property_reference_temp,
            'values': values_temp
        })
    elif(measure_name == 'humidity'):
        property_values.append({
            'entityPropertyReference': entity_property_reference_hum,
            'values': values_hum
        })
    LOGGER.info("property_values: %s", property_values)

    # marshall propertyValues and nextToken into final response
    return_obj = {
       'propertyValues': property_values,
       'nextToken': next_token
       }

    return return_obj

上記ソースコードについての補足としては、
・環境変数として、TIMESTREAM_DATABASE_NAMEとTIMESTREAM_TABLE_NAMEは参照するTimestreamのデータベース名とテーブル名を設定済み

・上記コードに2か所存在する
entity_property_reference_temp['entityId'] = 'xxx'
には、Twinmakerで作成したエンティティのEntityIdを実際は入力しております。

AWSの使い方があまりわかっていないので、解決するために必要な情報が抜けているかもしれないです。

解決策をご教授願います。

‐‐‐‐‐‐‐‐‐‐追記‐‐‐‐‐‐‐‐‐‐
こちらの行に関するエラーの原因がわかりました。'が全角の’になっていたため発生しておりました。
その部分を修正した結果、新たなエラーが発生しました。

[ERROR]	2022-10-20T07:20:45.195Z	400075f2-8527-4e3b-abba-7b794f2eb1fc	Exception while running query: An error occurred (ValidationException) when calling the Query operation: line 1:28: Column 'measure_value::bigint' does not exist

[ERROR] ValidationException: An error occurred (ValidationException) when calling the Query operation: line 1:28: Column 'measure_value::bigint' does not exist
Traceback (most recent call last):
  File "/var/task/lambda_function.py", line 109, in lambda_handler
    raise err
  File "/var/task/lambda_function.py", line 105, in lambda_handler
    QueryString = query_string
  File "/var/runtime/botocore/client.py", line 391, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/var/runtime/botocore/client.py", line 719, in _make_api_call
    raise error_class(parsed_response, operation_name)

Answer 1

解決済み
SQL辺りがいろいろ間違っている