Para proyectos nuevos, le recomendamos que utilice el nuevo servicio gestionado para Apache Flink Studio en lugar de Kinesis Data Analytics SQL for Applications. El servicio gestionado para Apache Flink Studio combina la facilidad de uso con capacidades analíticas avanzadas, lo que le permite crear aplicaciones sofisticadas de procesamiento de flujos en cuestión de minutos.

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Procesamiento previo de registros con una función de Lambda

Si los datos de su transmisión necesitan conversión de formato, transformación, enriquecimiento o filtrado, puede preprocesar los datos mediante una función. AWS Lambda Puede hacerlo antes de que se ejecute el código SQL de la aplicación o antes de que la aplicación cree un esquema a partir deel flujo de datos.

El uso de una función de Lambda para el procesamiento previo de registros es útil en las siguientes situaciones:

Uso de una función de Lambda para procesamiento previo de registros

Al crear su aplicación de Kinesis Data Analytics, habilita el procesamiento previo de en la página Conectar con una fuente.

Cuando se elige o se crea una función de Lambda para el procesamiento previo de los registros, estos se procesan antes de que se ejecute el código SQL de la aplicación o de que la aplicación genere un esquema a partir de ellos.

Permisos de procesamiento previo de Lambda

Para utilizar el procesamiento previo de Lambda, el rol de IAM de la aplicación requiere la política de permisos siguiente:

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

Métricas de procesamiento previo de Lambda

Puede utilizar Amazon CloudWatch para supervisar el número de invocaciones a Lambda, los bytes procesados, los éxitos y los errores, etc. Para obtener información sobre CloudWatch las métricas emitidas por el preprocesamiento de Kinesis Data Analytics Lambda, consulte Amazon Kinesis Analytics Metrics.

Uso AWS Lambda con la biblioteca de Kinesis Producer

Kinesis Producer Library (KPL) agrega pequeños registros formateados por el usuario en registros de mayor tamaño, hasta 1 MB, para utilizar mejor el rendimiento de Amazon Kinesis Data Streams. La Kinesis Client Library (KCL) para Java es compatible con la desagrupación de estos registros. Sin embargo, debe utilizar un módulo especial para desagregar los registros cuando los utilice AWS Lambda como consumidor de sus transmisiones.

Para obtener el código y las instrucciones del proyecto necesarios, consulte los módulos de desagregación de la biblioteca de productores de Kinesis para obtener más información. AWS LambdaGitHub Puede usar los componentes de este proyecto para procesar datos serializados de KPL AWS Lambda en Java, Node.js y Python. Estos componentes también se pueden utilizar como parte de una aplicación KCL multilenguaje.

Modelo de datos de entrada de eventos/modelo de respuesta de registros para el procesamiento previo de datos

Para realizar el procesamiento previo de los registros, la función de Lambda debe respetar los modelos necesarios de datos de entrada de eventos y de respuesta de registros.

Modelo de datos de entrada de eventos

Kinesis Data Analytics lee continuamente los datos de la transmisión de datos de Kinesis o la transmisión de entrega de Firehose. Para cada lote de registros que recupera, el servicio administra cómo se transfiere cada lote a su función de Lambda. Su función recibe una lista de registros como entrada. Dentro de su función, itera a través de la lista y aplica su lógica de negocio para llevar a cabo sus requisitos de procesamiento previo (tales como el enriquecimiento o la conversión de formato de datos).

El modelo de entrada de la función de preprocesamiento varía ligeramente en función de si los datos se recibieron de una transmisión de datos de Kinesis o de una transmisión de entrega de Firehose.

Si la fuente es un flujo de entrega de Firehose, el modelo de datos de entrada del evento es el siguiente:

Modelo de datos de solicitud de Kinesis Data Firehose

En el siguiente ejemplo, se muestra la entrada de una secuencia de entrega de 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 el origen es un flujo de datos de Kinesis, el modelo de datos de entrada de eventos es el siguiente:

Modelo de datos de solicitud de secuencias de Kinesis.

En el siguiente ejemplo, se muestra la entrada de un flujo de datos de 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
         }
    }
  ]
}

Modelo de respuesta de registros

Todos los registros que devuelve su función de procesamiento previo de Lambda (con ID de registro) que se envían a la función de Lambda se deben devolver. Deben contener los siguientes parámetros, de lo contrario, Kinesis Data Analytics los rechaza y los trata como error de procesamiento previo de datos. La parte de carga de datos del registro se puede transformar para cumplir los requisitos de procesamiento previo.

Modelo de datos de respuesta

En el siguiente ejemplo, se muestra el resultado de una función Lambda:

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

Errores comunes de procesamiento previo de datos

A continuación se indican los motivos habituales por los que un procesamiento previo puede generar un error.

Para errores de procesamiento previo de datos, Kinesis Data Analytics sigue reintentando las invocaciones de Lambda en el mismo conjunto de registros hasta que tiene éxito. Puede supervisar las siguientes CloudWatch métricas para obtener información sobre los fallos.