Pour les nouveaux projets, nous vous recommandons d'utiliser le nouveau service géré pour Apache Flink Studio plutôt que Kinesis Data Analytics SQL for Applications. Le service géré pour Apache Flink Studio allie facilité d’utilisation et capacités analytiques avancées, ce qui vous permet de créer des applications sophistiquées de traitement des flux en quelques minutes.

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Prétraitement des données à l’aide d’une fonction Lambda

Si les données de votre flux nécessitent une conversion de format, une transformation, un enrichissement ou un filtrage, vous pouvez les prétraiter à l'aide d'une AWS Lambda fonction. Vous pouvez effectuer cette opération avant que le code SQL de votre application s'exécute ou avant que votre application crée un schéma à partir de votre flux de données.

L’utilisation d’une fonction Lambda de prétraitement des enregistrements est utile dans les cas suivants :

L'utilisation d'une fonction Lambda pour le prétraitement des enregistrements

Lorsque vous créez votre application Kinesis Data Analytics, vous activez le prétraitement Lambda sur la page Se connecter à une source.

Lorsque vous choisissez ou créez une fonction Lambda pour le prétraitement d’enregistrements, les enregistrements sont prétraités avant que le code SQL de votre application ne s’exécute ou que votre application ne génère un schéma à partir des enregistrements.

Autorisations de prétraitement Lambda

Pour utiliser le prétraitement Lambda, le rôle IAM de l’application a besoin de la stratégie d’autorisations suivante :

     {
       "Sid": "UseLambdaFunction",
       "Effect": "Allow",
       "Action": [
           "lambda:InvokeFunction",
           "lambda:GetFunctionConfiguration"
       ],
       "Resource": "<FunctionARN>"
   }

Métriques de prétraitement Lambda

Vous pouvez utiliser Amazon CloudWatch pour surveiller le nombre d'appels Lambda, le nombre d'octets traités, les réussites et les échecs, etc. Pour plus d'informations sur CloudWatch les métriques émises par le prétraitement Lambda de Kinesis Data Analytics, consultez Amazon Kinesis Analytics Metrics.

Utilisation AWS Lambda avec la bibliothèque Kinesis Producer

La bibliothèque producteur Kinesis (KPL) regroupe de petits enregistrements formatés par l'utilisateur en enregistrements plus volumineux allant jusqu'à 1 Mo afin de mieux utiliser le débit Amazon Kinesis Data Streams. La bibliothèque Kinesis Client Library (KCL) pour Java prend en charge la désagrégation de ces enregistrements. Cependant, vous devez utiliser un module spécial pour désagréger les enregistrements lorsque vous les utilisez AWS Lambda en tant que consommateur de vos streams.

Pour obtenir le code de projet et les instructions nécessaires, consultez les modules de désagrégation de la bibliothèque Kinesis Producer pour plus d'informations. AWS LambdaGitHub Vous pouvez utiliser les composants de ce projet pour traiter des données sérialisées KPL AWS Lambda dans Java, Node.js et Python. Vous pouvez également utiliser ces composants dans le cadre d’une application KCL multi-lang.

Modèle de données d'entrée d'événement modèle de réponse d'enregistrement pour le prétraitement des données

Pour prétraiter des enregistrements, votre fonction Lambda doit être conforme aux modèles de données d’entrée d’événement et de réponse d’enregistrement imposés.

Modèle de données d'entrée d'événement

Kinesis Data Analytics lit en permanence les données de votre flux de données Kinesis ou de votre flux de diffusion Firehose. Pour chaque lot d’enregistrements qu’il récupère, le service gère la manière dont chaque lot est transmis à votre fonction Lambda. Votre fonction reçoit une liste d'enregistrements en entrée. Au sein de votre fonction, vous effectuez une itération dans la liste et vous appliquez votre logique métier pour réaliser vos exigences de prétraitement (par exemple, conversion du format de données ou enrichissement).

Le modèle d'entrée de votre fonction de prétraitement varie légèrement selon que les données proviennent d'un flux de données Kinesis ou d'un flux de diffusion Firehose.

Si la source est un flux de diffusion Firehose, le modèle de données d'entrée des événements est le suivant :

Modèle de données de demande Kinesis Data Firehose

L'exemple suivant montre l'entrée d'un flux de diffusion Firehose :

{
   "invocationId":"00540a87-5050-496a-84e4-e7d92bbaf5e2",
   "applicationArn":"arn:aws:kinesisanalytics:us-east-1:12345678911:application/lambda-test",
   "streamArn":"arn:aws:firehose:us-east-1:AAAAAAAAAAAA:deliverystream/lambda-test",
   "records":[
      {
         "recordId":"49572672223665514422805246926656954630972486059535892482",
         "data":"aGVsbG8gd29ybGQ=",
         "kinesisFirehoseRecordMetadata":{
            "approximateArrivalTimestamp":1520280173
         }
      }
   ]
}

Si la source est un flux de données Kinesis, le modèle de données d’entrée d’événement se présente comme suit :

Modèle de données de demande de flux Kinesis.

L'exemple suivant montre l'entrée d'un flux de données Kinesis :

{
  "invocationId": "00540a87-5050-496a-84e4-e7d92bbaf5e2",
  "applicationArn": "arn:aws:kinesisanalytics:us-east-1:12345678911:application/lambda-test",
  "streamArn": "arn:aws:kinesis:us-east-1:AAAAAAAAAAAA:stream/lambda-test",
  "records": [
    {
      "recordId": "49572672223665514422805246926656954630972486059535892482",
      "data": "aGVsbG8gd29ybGQ=",
      "kinesisStreamRecordMetadata":{
            "shardId" :"shardId-000000000003",
            "partitionKey":"7400791606",
            "sequenceNumber":"49572672223665514422805246926656954630972486059535892482",
            "approximateArrivalTimestamp":1520280173
         }
    }
  ]
}

Modèle de réponse d'enregistrement

Tous les enregistrements renvoyés à partir de votre fonction de prétraitement Lambda (avec les ID d'enregistrement) qui sont envoyés à la fonction Lambda doit être renvoyés. Ils doivent contenir les paramètres suivants. Sinon, Kinesis Data Analytics les rejette et les traite comme un échec de prétraitement des données. La partie charge utile des données de l'enregistrement peut être transformée pour réaliser les exigences de prétraitement.

Modèle de données de réponse

L'exemple suivant montre la sortie d'une fonction Lambda :

{
  "records": [
    {
      "recordId": "49572672223665514422805246926656954630972486059535892482",
      "result": "Ok",
      "data": "SEVMTE8gV09STEQ="
    }
  ]
}

Échecs courants du prétraitement des données

Voici les raisons courantes pour lesquelles le prétraitement peut échouer.

En cas d’échec de prétraitement des données, Kinesis Data Analytics continue de relancer les invocations Lambda sur le même ensemble d’enregistrements jusqu’à ce que cela aboutisse. Vous pouvez surveiller les CloudWatch indicateurs suivants pour mieux comprendre les défaillances.