Funzione Lambda (esempio 1)Funzione Lambda (esempio 2)

AWS Lambda attività

È possibile utilizzare un'lambdaattività per eseguire elaborazioni complesse sui messaggi. Ad esempio, puoi arricchire i messaggi con i dati provenienti dall'output di operazioni API esterne o filtrare i messaggi in base alla logica di Amazon DynamoDB. Tuttavia, non puoi utilizzare questa attività di pipeline per aggiungere altri messaggi o rimuovere messaggi esistenti prima di entrare in un data store.

La AWS Lambda funzione utilizzata in un'lambdaattività deve ricevere e restituire una matrice di oggetti JSON. Per vedere un esempio, consulta Funzione Lambda (esempio 1).

Per concedere AWS IoT Analytics l'autorizzazione a richiamare la funzione Lambda, è necessario aggiungere una policy. Ad esempio, esegui il seguente comando CLI e sostituiscilo exampleFunctionName con il nome della tua funzione Lambda, sostituiscilo 123456789012 con il tuo ID AWS account e usa l'Amazon Resource Name (ARN) della pipeline che richiama la funzione Lambda specificata.


aws lambda add-permission --function-name exampleFunctionName --action lambda:InvokeFunction --statement-id iotanalytics --principal iotanalytics.amazonaws.com --source-account 123456789012 --source-arn arn:aws:iotanalytics:us-east-1:123456789012:pipeline/examplePipeline

Il comando restituisce quanto segue:


{
    "Statement": "{\"Sid\":\"iotanalyticsa\",\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"iotanalytics.amazonaws.com\"},\"Action\":\"lambda:InvokeFunction\",\"Resource\":\"arn:aws:lambda:aws-region:aws-account:function:exampleFunctionName\",\"Condition\":{\"StringEquals\":{\"AWS:SourceAccount\":\"123456789012\"},\"ArnLike\":{\"AWS:SourceArn\":\"arn:aws:iotanalytics:us-east-1:123456789012:pipeline/examplePipeline\"}}}"
}

Per ulteriori informazioni, consulta Uso delle policy basate sulle risorse per AWS Lambda nella Guida per gli sviluppatori di AWS Lambda .

Funzione Lambda (esempio 1)

In questo esempio, la funzione Lambda aggiunge informazioni basate sui dati del messaggio originale. Un dispositivo pubblica un messaggio con un payload simile all'esempio seguente.


{
  "thingid": "00001234abcd",
  "temperature": 26,
  "humidity": 29,
  "location": {
    "lat": 52.4332935,
    "lon": 13.231694
  },
  "ip": "192.168.178.54",
  "datetime": "2018-02-15T07:06:01"
}

E il dispositivo ha la seguente definizione di pipeline.


{
    "pipeline": {
        "activities": [
            {
                "channel": {
                    "channelName": "foobar_channel",
                    "name": "foobar_channel_activity",
                    "next": "lambda_foobar_activity"
                }
            },
            {
                "lambda": {
                    "lambdaName": "MyAnalyticsLambdaFunction",
                    "batchSize": 5,
                    "name": "lambda_foobar_activity",
                    "next": "foobar_store_activity"
                }
            },
            {
                "datastore": {
                    "datastoreName": "foobar_datastore",
                    "name": "foobar_store_activity"
                }
            }
        ],
        "name": "foobar_pipeline",
        "arn": "arn:aws:iotanalytics:eu-west-1:123456789012:pipeline/foobar_pipeline"
    }
}

La seguente funzione Lambda Python (MyAnalyticsLambdaFunction) aggiunge GMaps l'URL e la temperatura, in gradi Fahrenheit, al messaggio.


import logging
import sys

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
streamHandler = logging.StreamHandler(stream=sys.stdout)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)

def c_to_f(c):
    return 9.0/5.0 * c + 32

def lambda_handler(event, context):
    logger.info("event before processing: {}".format(event))
    maps_url = 'N/A'

    for e in event:
        #e['foo'] = 'addedByLambda'
        if 'location' in e:
            lat = e['location']['lat']
            lon = e['location']['lon']
            maps_url = "http://maps.google.com/maps?q={},{}".format(lat,lon)

        if 'temperature' in e:
            e['temperature_f'] = c_to_f(e['temperature'])

        logger.info("maps_url: {}".format(maps_url))
        e['maps_url'] = maps_url

    logger.info("event after processing: {}".format(event))

    return event

Funzione Lambda (esempio 2)

Una tecnica utile consiste nel comprimere e serializzare i payload di messaggio, in modo da ridurre i costi di trasporto e storage. In questo secondo esempio, la funzione Lambda presuppone che il payload del messaggio rappresenti un originale JSON, che è stato compresso e quindi codificato in base64 (serializzato) come stringa. Restituisce il codice JSON originale.


import base64
import gzip
import json
import logging
import sys

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
streamHandler = logging.StreamHandler(stream=sys.stdout)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)

def decode_to_bytes(e):
    return base64.b64decode(e)

def decompress_to_string(binary_data):
    return gzip.decompress(binary_data).decode('utf-8')

def lambda_handler(event, context):
    logger.info("event before processing: {}".format(event))

    decompressed_data = []

    for e in event:
        binary_data = decode_to_bytes(e)
        decompressed_string = decompress_to_string(binary_data)

        decompressed_data.append(json.loads(decompressed_string))

    logger.info("event after processing: {}".format(decompressed_data))

    return decompressed_data

Avvertimento JavaScript è disabilitato o non è disponibile nel tuo browser.

Per usare la documentazione AWS, JavaScript deve essere abilitato. Consulta le pagine della guida del browser per le istruzioni.

Convenzioni dei documenti

Attività del datastore

AddAttributes attività