Comprensión profunda del almacenamiento de registros de Kafka (V)

Hay cinco archivos de puntos de control en el directorio raíz: clean-offset-checkpoint, log-start-offset-checkpoint, meta.properties, recovery-point-offset-check point y replication- punto de control de compensación.

Hay los siguientes directorios bajo el directorio de partición: 0000xxx.index (el desplazamiento tiene una longitud de 64 bits, la longitud se fija en 20 bits), 0000xxx.log, 0000xxx.timeindex.

También es posible que contenga archivos temporales, como .elited.cleaned.swap, y también puede contener una instantánea del índice txn.

Evolución del formato de registro 5.2

Versión 5.2.1 v0

Antes de Kafka 0.10.0

RECORD_OVERHEAD incluye desplazamiento (8B) y tamaño del mensaje (4B).

Los registros incluyen:

Crc32(4B): valor de verificación de crc32

Magic(1B): Número de versión del mensaje 0.

Atributos (1B): Atributos del mensaje. Los 3 bits inferiores indican el tipo de compresión: 0-ninguno 1-gzip2-snappy 3-lz4 (introducido en 0.9.x).

Longitud de la clave (4B): Indica la longitud de la clave del mensaje. -1 significa vacío.

Clave: Opcional

Valor de longitud (4B): La longitud del cuerpo real del mensaje. -1 significa vacío.

Valor: texto del mensaje. Puede estar vacío, como un mensaje de lápida.

Versión 5.2.2 v1

Kafka 0.1.0-0.1.0

Hay un campo de marca de tiempo (8B) mayor que v0, lo que indica que el mensaje marca de tiempo.

También se utiliza el cuarto bit del atributo. 0 indica que el tipo de marca de tiempo es CreateTime, 1 indica que el tipo de marca de tiempo es LogAppendTime.

El tipo de marca de tiempo se configura mediante el parámetro del lado del agente log.message.timestamp.type. El valor predeterminado es CreateTime, que utiliza la marca de tiempo creada por el productor.

Compresión de mensajes

Para garantizar la compresión de un extremo a otro, el servidor está configurado con compresión.tipo, que de forma predeterminada es "productor", lo que significa que el método de compresión utilizado por el productor se conserva. También se puede configurar como "gzip", "Snappy" o "LZ4".

Se comprimen varios mensajes en campos de valor para mejorar la relación de compresión.

Campo de longitud variable

Varints: El bit más alto de cada byte tiene un bit más significativo Excepto el último byte que es 1, el resto son cero. en orden inverso.

Para que la codificación sea más eficiente, Varints utiliza codificación en zigzag: sint32 corresponde a (n

versión 5 . 2 . 5 V2

Registrar lote

Primer desplazamiento:

Longitud:

Epoca principal de la partición:

Magia: fijada en 2

Atributos: dos bytes Los 3 bits inferiores indican el formato de compresión, el cuarto bit indica el tipo de marca de tiempo, el quinto bit indica la transacción (0-sin transacción 1-transacción) y el sexto bit indica el mensaje de control (0-). control no control 1).

Primera marca de tiempo:

Marca de tiempo máxima:

Identificación del productor:

Epoca del productor:

Primera secuencia:

Recuento de registros:

El mensaje de la versión v2 elimina el campo crc, aumenta la longitud (longitud total del mensaje), incremento de marca de tiempo (Incremento de marca de tiempo ), incremento de desplazamiento (incremento de desplazamiento) e información de encabezado, y estos atributos se descartan.

Registro

Longitud:

Atributo: obsoleto, pero aún representa 1B.

Incremento de marca de tiempo:

Incremento de compensación:

Título:

5.3 Índice de registro

Índice disperso: Siempre que se escribe una determinada cantidad (especificada por el parámetro del corredor log.index.interval.bytes, el valor predeterminado es 4096B), el archivo de índice de compensación y el archivo de índice de tiempo corresponden cada uno a una entrada de índice.

Estrategia de segmentación de registros:

1. El tamaño excede el valor configurado por el parámetro del lado del agente log.segment.bytes. El valor predeterminado es 1073741824 (1 GB).

2. La diferencia entre la marca de tiempo máxima del mensaje del segmento de registro actual y la marca de tiempo actual del sistema es mayor que log.roll.ms o log.roll.hours ms tiene una prioridad más alta y el valor predeterminado. es log.roll.hours=168(7 días).

3. El tamaño del archivo de índice o del archivo de índice de marca de tiempo es mayor que el valor configurado por log.index.size.max.bytes. El valor predeterminado es 10485760 (100 MB).

4. Desplazamiento - desplazamiento básico>: entero. Valor máximo

5.3.1 Índice de desplazamiento

Cada elemento del índice ocupa 8 bytes y se divide en dos partes: 1. desplazamiento relativo (4B) 2. dirección física de posición (4B).

Utilice el script kafka-dump-log.sh para el análisis. Archivos de índice (incluido índice de tiempo, instantánea, txnindex, etc.), de la siguiente manera:

bin/Kafka-dump-log sh-files/tmp/Kafka-logs/topic id-0/00... …00 . index

Si el parámetro del lado del agente log.index.size.max.bytes no es un múltiplo de 8, se convertirá automáticamente a un múltiplo de 8 internamente.

Índice de marca de tiempo

Cada entrada de índice ocupa 12 bytes y se divide en dos partes: 1. Marca de tiempo La marca de tiempo máxima del segmento de registro actual (12b) 2. El desplazamiento relativo corresponde a la marca de tiempo del desplazamiento relativo (4B).

Si el parámetro del lado del corredor log.index.size.max.bytes no es un múltiplo de 12, se convertirá automáticamente a un múltiplo de 12 internamente.

5.4 Limpieza de registros

Las políticas de limpieza de registros se pueden controlar a nivel de tema.

5.4.1 Eliminación de registros

El parámetro del lado del agente log.cleanup.policy está configurado para eliminar (el valor predeterminado es eliminar).

Registro de retención de parámetros de fin del ciclo de detección.

1. Basado en el tiempo

Parámetros del corredor log.retention.hours, log.retention.minutos, log.retention.ms, prioridad ms>minutos>hours

Cuando se elimina, sufijo. Agregue eliminar primero y retrase la eliminación según la configuración de file.delete.delay.ms (predeterminado 60000).

2. Según el tamaño del registro

El tamaño total del registro es el parámetro del lado del agente log.retention.bytes (el valor predeterminado es -1, que indica infinito).

El tamaño del segmento de registro es el parámetro del lado del agente log.segment.bytes (el valor predeterminado es 1073741824, 1 GB).

3. DeleteRecord() basado en el desplazamiento inicial del registro

Solicitud DeleteRecordRequest

1.

KafkaAdminClient

2.script kafka-delete-record.sh

Compresión de registros

El parámetro del lado del agente log.cleanup.policy está configurado en compacto, registro .cleaner. enable está establecido en verdadero (el valor predeterminado es verdadero).

Almacenamiento en disco 5.5

Pruebas relacionadas: una pila aleatoria de discos forma 6 matrices RAID-5 a 7200 r/min, escritura lineal de 600 MB/s, escritura aleatoria de 100 KB/s, escritura aleatoria de 400 MB /s, memoria lineal 3,6 GB/s.

5.5.1 Caché de página

El parámetro vm.dirty_background_ratio del sistema operativo Linux se utiliza para especificar que pdflush/flush/kdmflush se activará cuando el número de páginas sucias alcance un porcentaje del sistema, que generalmente es inferior a 10. No se recomienda 0.

Vm.dirty_ratio significa que el disco se limpia después de alcanzar el porcentaje de páginas sucias, pero se bloquean las nuevas solicitudes de io.

Kafka también proporciona funciones de vaciado de disco sincrónico y vaciado de disco forzado intermitente (fsync), que se pueden controlar mediante parámetros como log.flush.interval.messages y log.flush.interval.ms

Kafka no recomienda el uso de particiones de intercambio. El límite superior del parámetro vm.swappiness es 100 y el límite inferior es 0. Se recomienda configurarlo en 1.

Proceso de entrada/salida del disco

Generalmente, hay cuatro situaciones de IO del disco:

1. El usuario llama a la biblioteca C estándar para realizar operaciones de IO. y el flujo de datos es: búfer de aplicación -> búfer io estándar de la biblioteca C -> El sistema de archivos también se almacena en caché -> A través de un sistema de archivos específico al disco

2. El usuario llama al archivo IO, los datos El flujo es: búfer de aplicación ->; El sistema de archivos también se almacena en caché ->; Al disco a través de un sistema de archivos específico

3. Cuando el usuario abre un archivo, use O_DIRECT para omitir el caché de la página y directamente. leer y escribir en el disco.

4. El usuario utiliza una herramienta similar a dd, utiliza el parámetro directo para omitir el caché del sistema y utiliza directamente el sistema de archivos para leer y escribir en el disco.

Existen cuatro estrategias de programación de IO en los sistemas Linux:

1. Sin cirugía

2. CFQ

3. p>

4. Ocurrencia temprana

Copia cero

se refiere a datos copiados directamente desde archivos de disco a dispositivos de tarjeta de red, sin necesidad de aplicaciones.

Para Linux, se basa en el sendfile() subyacente.

Para Java, la implementación subyacente de FileChannal.transferTo() es sendfile().