Escrivir en Amazon Kinesis Data Streams mediante el agente de Kinesis

El agente de Kinesis es una aplicación de software Java independiente que ofrece una forma sencilla de recopilar y enviar datos a Kinesis Data Streams. El agente monitoriza constantemente un conjunto de archivos y envía nuevos datos a su secuencia. El agente se encarga de rotar los archivos, crear puntos de restauración y realizar reintentos cuando se producen errores. El agente entrega todos los datos de manera confiable, puntual y sencilla. También emite CloudWatch métricas de Amazon para ayudarte a supervisar y solucionar mejor los problemas del proceso de streaming.

De forma predeterminada, los registros de cada archivo se analizan en función del carácter de nueva línea ('\n'). Sin embargo, el agente también se puede configurar para analizar registros multilínea (consulte Especificar las opciones de configuración del agente).

Puede instalar el agente en entornos de servidor basados en Linux, como servidores web, de registro o de base de datos. Después de instalar el agente, configúrelo especificando los archivos que desee monitorizar y la secuencia de los datos. Una vez configurado, el agente recopila datos de los archivos de forma duradera y los envía de forma confiable a la secuencia.

Cumplir con los requisitos previos para el agente de Kinesis

Descargar e instalar el agente

Primero, conéctese a la instancia. Para obtener más información, consulte Connect to Your Instance en la Guía del EC2 usuario de Amazon. Si tiene problemas para conectarse, consulte Solución de problemas de conexión a su instancia en la Guía del EC2 usuario de Amazon.

Para configurar el agente a través de la AMI de Amazon Linux

Use el siguiente comando para descargar e instalar el agente:

sudo yum install –y aws-kinesis-agent

Para configurar el agente en Red Hat Enterprise Linux

Use el siguiente comando para descargar e instalar el agente:

sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn2.noarch.rpm

Configuración del agente en un contenedor de Docker

El agente de Kinesis también puede ejecutarse en un contenedor además de a través de la base de contenedores amazonlinux. Utilice el siguiente Dockerfile y ejecute docker build.

FROM amazonlinux

RUN yum install -y aws-kinesis-agent which findutils
COPY agent.json /etc/aws-kinesis/agent.json

CMD ["start-aws-kinesis-agent"]

Configurar e iniciar el agente

El agente ya está se ejecutando como un servicio de sistema en segundo plano. Monitoriza constantemente los archivos especificados y envía datos a la secuencia especificada. La auditoría de actividad se registra en /var/log/aws-kinesis-agent/aws-kinesis-agent.log.

Especificar las opciones de configuración del agente

El agente admite las dos opciones de configuración obligatorias, filePattern y kinesisStream, además de configuraciones opcionales para activar características adicionales. Las opciones de configuración obligatorias y opcionales se especifican en /etc/aws-kinesis/agent.json.

Cada vez que cambie el archivo de configuración, debe detener y comenzar el agente con los siguientes comandos:

sudo service aws-kinesis-agent stop
sudo service aws-kinesis-agent start

También puede hacerlo con el siguiente comando:

sudo service aws-kinesis-agent restart

Las opciones de configuración generales son las siguientes.

Las opciones de configuración de flujo son las siguientes.

Monitorear varios directorios de archivos y escribir en varias secuencias

Puede configurar el agente para que monitorice varios directorios de archivos y envíe datos a varias secuencias especificando varias opciones de configuración de secuencia. En el siguiente ejemplo de configuración, el agente supervisa dos directorios de archivos y envía datos a un flujo de Kinesis y a un flujo de entrega de Firehose, respectivamente. Tenga en cuenta que puede especificar diferentes puntos de conexión para Kinesis Data Streams y Firehose para que su flujo de Kinesis y su flujo de entrega de Firehose no tengan que estar en la misma región.

{
    "cloudwatch.emitMetrics": true,
    "kinesis.endpoint": "https://your/kinesis/endpoint", 
    "firehose.endpoint": "https://your/firehose/endpoint", 
    "flows": [
        {
            "filePattern": "/tmp/app1.log*", 
            "kinesisStream": "yourkinesisstream"
        }, 
        {
            "filePattern": "/tmp/app2.log*",
            "deliveryStream": "yourfirehosedeliverystream" 
        }
    ] 
}

Para más información sobre el uso del agente con Firehose, consulte Escritura en Amazon Data Firehose con el agente de Kinesis.

Usar el agente para preprocesar el agente

El agente puede preprocesar los registros previamente analizados de los archivos monitorizados antes de enviarlos a la secuencia. Para habilitar esta característica, añada la opción de configuración dataProcessingOptions al flujo de archivos. Se pueden agregar una o más opciones de procesamiento, que se ejecutarán en el orden especificado.

El agente es compatible con las siguientes opciones de procesamiento. Al ser de código abierto, puede desarrollar y ampliar las opciones de procesamiento del agente. Puede descargar el agente desde el agente de Kinesis.

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG"
}

64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291

{"host":"64.242.88.10","ident":null,"authuser":null,"datetime":"07/Mar/2004:16:10:02 -0800","request":"GET /mailman/listinfo/hsdivision HTTP/1.1","response":"200","bytes":"6291"}

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG",
    "customFieldNames": ["f1", "f2", "f3", "f4", "f5", "f6", "f7"]
}

{"f1":"64.242.88.10","f2":null,"f3":null,"f4":"07/Mar/2004:16:10:02 -0800","f5":"GET /mailman/listinfo/hsdivision HTTP/1.1","f6":"200","f7":"6291"}

{ 
    "flows": [
        {
            "filePattern": "/tmp/app.log*", 
            "kinesisStream": "my-stream",
            "dataProcessingOptions": [
                {
                    "optionName": "LOGTOJSON",
                    "logFormat": "COMMONAPACHELOG"
                }
            ]
        }
    ] 
}

{ 
    "flows": [
        {
            "filePattern": "/tmp/app.log*", 
            "kinesisStream": "my-stream",
            "multiLineStartPattern": "\\[SEQUENCE=",
            "dataProcessingOptions": [
                {
                    "optionName": "SINGLELINE"
                },
                {
                    "optionName": "CSVTOJSON",
                    "customFieldNames": [ "field1", "field2", "field3" ],
                    "delimiter": "\\t"
                }
            ]
        }
    ] 
}

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG",
    "matchPattern": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3})",
    "customFieldNames": ["host", "ident", "authuser", "datetime", "request", "response"]
}

123.45.67.89 - - [27/Oct/2000:09:27:09 -0400] "GET /java/javaResources.html HTTP/1.0" 200

{"host":"123.45.67.89","ident":null,"authuser":null,"datetime":"27/Oct/2000:09:27:09 -0400","request":"GET /java/javaResources.html HTTP/1.0","response":"200"}

Usar los comandos de la CLI

Iniciar automáticamente al agente al arrancar el sistema:

sudo chkconfig aws-kinesis-agent on

Compruebe el estado del agente:

sudo service aws-kinesis-agent status

Detener el agente:

sudo service aws-kinesis-agent stop

Leer el archivo de registro del agente desde esta ubicación:

/var/log/aws-kinesis-agent/aws-kinesis-agent.log

Desinstalar el agente:

sudo yum remove aws-kinesis-agent

Preguntas frecuentes

¿Hay un agente de Kinesis para Windows?

El agente de Kinesis para Windows es un software diferente del agente de Kinesis para plataformas Linux.

¿Por qué se ralentiza el agente de Kinesis o aumenta el valor de RecordSendErrors?

Normalmente esto se debe a la limitación de Kinesis. Compruebe la métrica WriteProvisionedThroughputExceeded para Kinesis Data Streams o la métrica ThrottledRecords para flujos de entrega de Firehose. Cualquier aumento a partir de 0 en estas métricas indica que es necesario aumentar los límites de flujo. Para obtener más información, consulte Límites de flujos de datos de Kinesis y flujos de entrega de Amazon Firehose.

Una vez que descarte la limitación de velocidad, compruebe si el agente de Kinesis está configurado para enviar a la cola una gran cantidad de archivos pequeños. Se produce un retraso cuando el agente de Kinesis sigue un archivo nuevo, por lo que el agente de Kinesis debería seguir una pequeña cantidad de archivos de mayor tamaño. Intente consolidar los archivos de registro en archivos más grandes.

¿Por qué se producen excepciones java.lang.OutOfMemoryError ?

El agente de Kinesis no tiene memoria suficiente para gestionar la carga de trabajo actual. Intente aumentar JAVA_START_HEAP y JAVA_MAX_HEAP en /usr/bin/start-aws-kinesis-agent y reinicie el agente.

¿Por qué se producen excepciones IllegalStateException : connection pool shut down?

El agente de Kinesis no tiene suficientes conexiones para gestionar la carga de trabajo actual. Intente aumentar maxConnections y maxSendingThreads en los ajustes generales de configuración del agente en /etc/aws-kinesis/agent.json. El valor predeterminado para estos campos es 12 veces los procesadores de tiempo de ejecución disponibles. Consulte AgentConfiguration.java para obtener más información sobre las configuraciones avanzadas de los agentes.

¿Cómo puedo depurar otro problema con el agente de Kinesis?

Los registros de nivel DEBUG pueden habilitarse en /etc/aws-kinesis/log4j.xml.

¿Cómo debo configurar el agente de Kinesis?

Cuanto menor sea el valor de maxBufferSizeBytes, más frecuentemente enviará datos el agente de Kinesis. Esto puede ser bueno ya que disminuye el tiempo de entrega de los registros, pero también aumenta las solicitudes por segundo a Kinesis.

¿Por qué el agente de Kinesis envía registros duplicados?

Esto ocurre debido a una mala configuración en el seguimiento de archivos. Asegúrese de que cada fileFlow’s filePattern solo coincida con un archivo. Esto también puede ocurrir si el modo logrotate que se está utilizando está en modo copytruncate. Intente cambiar al modo predeterminado o al de creación para evitar la duplicación. Para obtener más información sobre la gestión de registros duplicados, consulte Handling Duplicate Records.