Configure Flink job log export - Realtime Compute for Apache Flink

Puede consultar los registros de trabajos en la página de información del trabajo en la consola de Flink. También puede exportarlos a un servicio de almacenamiento externo como OSS, SLS o Kafka y configurar el nivel de registro para la salida. A continuación se explica cómo hacerlo.

Notas

Después de configurar la exportación de registros a OSS, SLS o Kafka, debe reiniciar el trabajo.
Si configura la exportación de registros hacia otro servicio de almacenamiento pero no deshabilita la función de archivado de registros, estos seguirán guardándose en OSS o en el almacenamiento totalmente administrado configurado para su espacio de trabajo. Si deshabilita el archivado de registros, ya no podrá consultar los registros de trabajos en la consola de Flink.

En la página deployment details, dentro de la sección log configuration, use el interruptor de la fila log archiving para habilitar o deshabilitar esta función.
Puede usar ${secret_values.xxxx} en la configuración de registros para hacer referencia a las variables almacenadas en la administración de variables. Para obtener más información, consulte project variable.

Configurar la exportación de registros de un solo trabajo

Consola

Vaya a la página de configuración de registros de un despliegue.
1. Inicie sesión en la consola de Realtime Compute for Apache Flink.
2. En la columna Actions del espacio de trabajo de destino, haga clic en Console.
3. En el panel de navegación izquierdo, haga clic en O&M > Deployments. Luego, haga clic en el nombre del despliegue de destino.
4. En la pestaña Configuration y dentro de la sección Logging, haga clic en Edit a la derecha.
5. Para Logging Profile, seleccione Custom Template.
Haga clic en Apply with System Profiles > default.
Haga clic en Add Appender y seleccione un destino.

Configure los parámetros del appender seleccionado.

Para enrutar los registros de distintos niveles a diferentes destinos, configure reglas de filtrado para el appender. Para obtener más información, consulte Configure log filtering based on log levels.

SLS

Parámetro	Descripción
name	Nombre personalizado para el appender.
type	Tipo de appender. Este valor es fijo y corresponde a `SLS`. No lo modifique.
pattern	Formato de salida de los registros. El valor predeterminado es `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`, que genera entradas de registro como `2023-10-01 14:23:45,678{GMT+8} INFO com.example.MyClass - This is a test log message`.
flushIntervalSeconds	Intervalo, en segundos, con el que los datos de registro se envían al destino.
flushIntervalEventCount	Número de entradas de registro que se agrupan antes de escribir en el destino de almacenamiento. Nota Si configura tanto `flushIntervalSeconds` como `flushIntervalEventCount`, la operación de escritura se activa cuando se alcanza cualquiera de los dos umbrales, el que ocurra primero.
authenticationMode	AccessKey SLS Token Nota Si selecciona SLS Token, solo podrá exportar registros a un logstore de un proyecto ubicado en la misma región que su espacio de trabajo de Flink. Esta función es compatible únicamente con VVR 11.5 y versiones posteriores.
project	Nombre del proyecto de SLS.
logStore	Nombre del logstore.
endpoint	Endpoint privado de la región del servicio SLS. Para obtener más información, consulte Endpoints.
accessKeyId	AccessKey ID y AccessKey Secret de la cuenta utilizada para el servicio SLS. Para saber cómo obtener un par de AccessKey, consulte Obtain an AccessKey pair. Para evitar los riesgos de seguridad de usar claves en texto plano, se recomienda almacenar los valores de AccessKey en variables de namespace. Para obtener más información, consulte Namespace variables. Nota Si sus servicios SLS y Realtime Compute for Apache Flink están en cuentas de Alibaba Cloud distintas, debe otorgar a la cuenta de Flink permiso para escribir en SLS. Para más detalles, consulte Create a custom RAM policy. A continuación se muestran ejemplos de políticas: Alcance de SLS sin restricciones `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:Get", "log:PostLogStoreLogs" ], "Resource": "" } ] }` Alcance limitado a un recurso específico de SLS `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:PostLogStoreLogs", "log:GetLogStore" ], "Resource": "acs:log:cn-beijing:152940222687****:project/test-vvp-sls/logstore/test-ltest" } ] }`
accessKeySecret

OSS

Los valores predeterminados de los parámetros numéricos son los siguientes: flushIntervalSeconds es 10, flushIntervalEventCount es 100 y rollingBytes es 1048576.

Parámetro	Descripción
name	Nombre personalizado para el appender.
type	Tipo de appender. Este valor es fijo y corresponde a `OSS`. No lo modifique.
pattern	Formato de salida de los registros. El valor predeterminado es `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`.
baseUri	Nombre del bucket de OSS.
endpoint	Endpoint de la región de su servicio OSS. Para obtener más información, consulte Regions and endpoints. Use el endpoint de la columna Endpoint (region) que corresponde a la fila Access from ECS over the VPC (internal network).
accessKeyId	AccessKey ID y AccessKey Secret de la cuenta utilizada para el servicio OSS. Para saber cómo obtener un par de AccessKey, consulte Obtain an AccessKey pair. Para evitar los riesgos de seguridad de usar claves en texto plano, se recomienda almacenar los valores de AccessKey en variables de namespace. Para obtener más información, consulte Namespace variables. Nota Estos parámetros son obligatorios solo si el bucket de OSS de destino pertenece a una cuenta de Alibaba Cloud distinta a la de su servicio Flink. Si ambos están en la misma cuenta, elimine estos parámetros de la configuración.
accessKeySecret
flushIntervalSeconds	Intervalo, en segundos, con el que los datos de registro se envían al destino.
flushIntervalEventCount	Número de entradas de registro que se agrupan antes de escribir en el destino de almacenamiento. Nota Si configura tanto `flushIntervalSeconds` como `flushIntervalEventCount`, la operación de escritura se activa cuando se alcanza cualquiera de los dos umbrales, el que ocurra primero.
rollingBytes	Tamaño máximo de un archivo de registro individual en OSS. Al alcanzar este tamaño, los datos posteriores se escriben en un nuevo archivo de registro.

Kafka

Nota

No son compatibles los clústeres de Kafka que tienen habilitada la autenticación Kerberos.

Requisitos previos

Dado que el cargador de clases de plugins de Flink carga el plugin KafkaAppender, debe declarar explícitamente su ruta de paquete en la configuración de Flink. Esto permite que la aplicación cargue el plugin para el despliegue actual.

En la página Deployments, haga clic en el nombre de su despliegue. En la pestaña Configuration y dentro de la sección Parameters, agregue el siguiente parámetro al campo Other Configuration.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Configuración

Parámetro	Descripción
name	Nombre personalizado para el appender.
type	Tipo de appender. Este valor es fijo y corresponde a `KafkaVVP`. No lo modifique.
pattern	Formato de salida de los registros. El valor predeterminado es `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`.
bootstrap.servers	Direcciones de los brokers de Kafka.
acks	Número de réplicas de partición que deben recibir un mensaje antes de que el productor considere correcta la escritura. Para obtener más información, consulte acks.
buffer.memory	Memoria total, en bytes, que el productor puede usar para almacenar registros en búfer.
retries	Número de veces que el productor reintentará enviar un mensaje tras un error transitorio.
compression.type	Tipo de compresión para los datos generados por el productor. Valores válidos: `none`, `gzip`, `snappy`, `lz4` y `zstd`.

Haga clic en Save.
En la parte superior de la página, haga clic en Start.

Método XML

Para configurar el registro de un despliegue, vaya a su página de configuración de registros.
1. Inicie sesión en la consola de Realtime Compute for Apache Flink.
2. En la columna Actions del espacio de trabajo de destino, haga clic en Console.
3. En el panel de navegación izquierdo, haga clic en O&M > Deployments y luego haga clic en el nombre del despliegue de destino.
4. En la pestaña Configuration y dentro de la sección Logging, haga clic en Edit.
5. Establezca Logging Profile en Custom Template.

Configure la salida de registros.

Según su destino de almacenamiento, copie la configuración correspondiente, péguela en el cuadro de entrada y actualice los parámetros con los datos de su almacenamiento. Para enrutar los registros de distintos niveles a diferentes destinos, puede configurar reglas de filtrado para cada appender. Para obtener más información, consulte Configure log outputs for different log levels.

OSS

<?xml version="1.0" encoding="UTF-8"?>
<Configuration xmlns="http://logging.apache.org/log4j/2.0/config" 
strict="true" packages="com.ververica.platform.logging.appender" status="WARN">  
  <Appenders> 
    <Appender name="StdOut" type="Console"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/> 
    </Appender> 
    <Appender name="RollingFile" type="RollingFile" fileName="${sys:log.file}" filePattern="${sys:log.file}.%i"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  
      <Policies> 
        <SizeBasedTriggeringPolicy size="20 MB"/> 
      </Policies>  
      <DefaultRolloverStrategy max="4"/> 
    </Appender>  
    <Appender name="OSS" type="OSS">
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  
      <!-- The final effective log path is: ${baseUri}/logs/${namespace}/${deploymentId}/{jobId}/ -->
      <Property name="namespace">{{ namespace }}</Property> <!-- Do not modify this line -->
      <Property name="baseUri">oss://YOUR-BUCKET-NAME/</Property>
      <Property name="endpoint">https://YOUR-ENDPOINT</Property> 
      <Property name="accessKeyId">${secret_values.accessKeyId}</Property>
      <Property name="secretAccessKey">${secret_values.accessKeySecret}</Property>
      <Property name="flushIntervalSeconds">10</Property>  
      <Property name="flushIntervalEventCount">100</Property>  
      <Property name="rollingBytes">10485760</Property>  
    </Appender>
   <Appender name="StdOutErrConsoleAppender" type="Console">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
   </Appender>
   <Appender name="StdOutFileAppender" type="RollingFile" fileName="${sys:stdout.file}" filePattern="${sys:stdout.file}.%i">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
     <Policies>
     <SizeBasedTriggeringPolicy size="1 GB"/>
     </Policies>
     <DefaultRolloverStrategy max="2"/>
   </Appender>
   <Appender name="StdErrFileAppender" type="RollingFile" fileName="${sys:stderr.file}" filePattern="${sys:stderr.file}.%i">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
     <Policies>
     <SizeBasedTriggeringPolicy size="1 GB"/>
     </Policies>
     <DefaultRolloverStrategy max="2"/>
   </Appender>
  </Appenders>  
  <Loggers> 
    <Logger level="INFO" name="org.apache.hadoop"/>  
    <Logger level="INFO" name="org.apache.kafka"/>  
    <Logger level="INFO" name="org.apache.zookeeper"/>  
    <Logger level="INFO" name="akka"/>  
    <Logger level="ERROR" name="org.jboss.netty.channel.DefaultChannelPipeline"/>  
    <Logger level="OFF" name="org.apache.flink.runtime.rest.handler.job.JobDetailsHandler"/> 
    <Logger level="ERROR" name="org.apache.flink.fs.osshadoop.shaded.com.aliyun.oss"/>
  <Logger level="INFO" name="StdOutErrRedirector.StdOut" additivity="false">
    <AppenderRef ref="StdOutFileAppender"/>
    <AppenderRef ref="StdOutErrConsoleAppender"/>
  </Logger>
  <Logger level="INFO" name="StdOutErrRedirector.StdErr" additivity="false">
    <AppenderRef ref="StdErrFileAppender"/>
    <AppenderRef ref="StdOutErrConsoleAppender"/>
  </Logger>
    {%- for name, level in userConfiguredLoggers -%} 
      <Logger level="{{ level }}" name="{{ name }}"/> 
    {%- endfor -%}
    <Root level="{{ rootLoggerLogLevel }}"> 
      <AppenderRef ref="StdOut"/>
      <AppenderRef ref="RollingFile"/>  
      <AppenderRef ref="OSS"/> 
    </Root>
  </Loggers> 
</Configuration>

Parámetro	Descripción
YOUR-BUCKET-NAME	Reemplace este valor por el nombre de su bucket de OSS.
YOUR-ENDPOINT	Reemplace este valor por su endpoint de OSS. Para obtener más información, consulte Regions and endpoints. Este es el valor de la columna Endpoint en la fila Access from ECS over the VPC (internal network).
YOUR-OSS-ACCESSKEYID	Use el AccessKey ID y el AccessKey secret de su cuenta de OSS. Para obtener más información sobre cómo conseguirlos, consulte Obtain an AccessKey pair. Para evitar exponer credenciales en texto plano, este ejemplo usa variables de namespace para especificar los valores de AccessKey. Para obtener más información, consulte Namespace variables. Nota Estos parámetros son obligatorios solo si el bucket de OSS y su servicio Realtime Compute for Apache Flink están en cuentas distintas. Si el bucket está en la misma cuenta, puede omitir estos parámetros.
YOUR-OSS-ACCESSKEYSECRET
flushIntervalSeconds	Intervalo, en segundos, con el que los datos de registro se escriben en el destino de almacenamiento.
flushIntervalEventCount	Número de entradas de registro que se almacenan en búfer antes de escribir los datos en el destino. Nota Si configura tanto `flushIntervalSeconds` como `flushIntervalEventCount`, la operación de escritura se activa cuando se alcanza cualquiera de los dos umbrales.
rollingBytes	Tamaño máximo de un archivo de registro individual en OSS. Cuando un archivo alcanza este tamaño, se crea uno nuevo para los registros posteriores.

Nota

En un trabajo JAR, el código de negocio se carga mediante un cargador de clases distinto al del sistema de registros de Flink. De forma predeterminada, el plugin de registros valida estrictamente los cargadores de clases (isInitStrictly tiene como valor predeterminado true), por lo que dichos registros se omiten y no se exportan a OSS. Para recopilar los registros que imprime el código de negocio en un trabajo JAR, agregue la siguiente configuración al appender de OSS:

<Property name="isInitStrictly">false</Property>

SLS

<?xml version="1.0" encoding="UTF-8"?>
<Configuration xmlns="http://logging.apache.org/log4j/2.0/config" 
strict="true" packages="com.ververica.platform.logging.appender" status="WARN">  
  <Appenders> 
    <Appender name="StdOut" type="Console"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/> 
    </Appender>  
    <Appender name="RollingFile" type="RollingFile" fileName="${sys:log.file}" filePattern="${sys:log.file}.%i"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  
      <Policies> 
        <SizeBasedTriggeringPolicy size="5 MB"/> 
      </Policies>  
      <DefaultRolloverStrategy max="1"/> 
    </Appender>  
    <Appender name="SLS" type="SLS">
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  
      <!-- The final effective log path is: ${baseUri}/logs/${namespace}/${deploymentId}/{jobId}/ -->
      <Property name="namespace">{{ namespace }}</Property> <!-- Do not modify this line -->
      <Property name="project">YOUR-SLS-PROJECT</Property>  
      <Property name="logStore">YOUR-SLS-LOGSTORE</Property> 
      <Property name="endpoint">YOUR-SLS-ENDPOINT</Property> 
      <Property name="accessKeyId">${secret_values.accessKeyId}</Property> 
      <Property name="accessKeySecret">${secret_values.accessKeySecret}</Property> 
      <Property name="topic">{{ namespace }}:{{ deploymentId }}:{{ jobId }}</Property>
      <Property name="deploymentName">{{ deploymentName }}</Property>
      <Property name="flushIntervalSeconds">10</Property>
      <Property name="flushIntervalEventCount">100</Property>
    </Appender>
   <Appender name="StdOutErrConsoleAppender" type="Console">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
   </Appender>
   <Appender name="StdOutFileAppender" type="RollingFile" fileName="${sys:stdout.file}" filePattern="${sys:stdout.file}.%i">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
     <Policies>
     <SizeBasedTriggeringPolicy size="1 GB"/>
     </Policies>
     <DefaultRolloverStrategy max="2"/>
   </Appender>
   <Appender name="StdErrFileAppender" type="RollingFile" fileName="${sys:stderr.file}" filePattern="${sys:stderr.file}.%i">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
     <Policies>
     <SizeBasedTriggeringPolicy size="1 GB"/>
     </Policies>
     <DefaultRolloverStrategy max="2"/>
   </Appender>
  </Appenders>  
  <Loggers> 
    <Logger level="INFO" name="org.apache.hadoop"/>  
    <Logger level="INFO" name="org.apache.kafka"/>  
    <Logger level="INFO" name="org.apache.zookeeper"/>  
    <Logger level="INFO" name="akka"/>  
    <Logger level="ERROR" name="org.jboss.netty.channel.DefaultChannelPipeline"/>  
    <Logger level="OFF" name="org.apache.flink.runtime.rest.handler.job.JobDetailsHandler"/> 
    <Logger level="ERROR" name="org.apache.flink.fs.osshadoop.shaded.com.aliyun.oss"/>
  <Logger level="INFO" name="StdOutErrRedirector.StdOut" additivity="false">
    <AppenderRef ref="StdOutFileAppender"/>
    <AppenderRef ref="StdOutErrConsoleAppender"/>
  </Logger>
  <Logger level="INFO" name="StdOutErrRedirector.StdErr" additivity="false">
    <AppenderRef ref="StdErrFileAppender"/>
    <AppenderRef ref="StdOutErrConsoleAppender"/>
  </Logger>
    {%- for name, level in userConfiguredLoggers -%} 
      <Logger level="{{ level }}" name="{{ name }}"/> 
    {%- endfor -%}
    <Root level="{{ rootLoggerLogLevel }}"> 
      <AppenderRef ref="StdOut"/>
      <AppenderRef ref="RollingFile"/>  
      <AppenderRef ref="SLS"/> 
    </Root>
  </Loggers> 
</Configuration>

Nota

Las propiedades namespace, deploymentId, jobId y deploymentName del código son variables de Twig. No las modifique; de lo contrario, el trabajo no podrá iniciarse.

Parámetro	Descripción
YOUR-SLS-PROJECT	Reemplace este valor por el nombre de su proyecto de SLS.
YOUR-SLS-LOGSTORE	Reemplace este valor por el nombre de su logstore de SLS.
YOUR-SLS-ENDPOINT	Reemplace este valor por el endpoint privado de la región de su servicio SLS. Para obtener más información, consulte Service endpoints.
YOUR-SLS-ACCESSKEYID	Use el AccessKey ID y el AccessKey secret de su cuenta de SLS. Para obtener más información sobre cómo conseguirlos, consulte Obtain an AccessKey pair. Para evitar exponer credenciales en texto plano, este ejemplo usa variables de namespace para especificar los valores de AccessKey. Para obtener más información, consulte Namespace variables. Nota Si sus servicios SLS y Realtime Compute for Apache Flink están en cuentas distintas, debe otorgar a la cuenta de Flink permisos de escritura en SLS. Para obtener más información, consulte Create a custom RAM policy. Use una de las siguientes políticas: Alcance de SLS sin restricciones `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:Get", "log:PostLogStoreLogs" ], "Resource": "" } ] }` Limitada a un recurso específico de SLS. Ejemplo: `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:PostLogStoreLogs", "log:GetLogStore" ], "Resource": "acs:log:cn-beijing:152940222687****:project/test-vvp-sls/logstore/test-ltest" } ] }`
YOUR-SLS-ACCESSKEYSECRET
flushIntervalSeconds	Intervalo, en segundos, con el que los datos de registro se escriben en el destino de almacenamiento.
flushIntervalEventCount	Número de entradas de registro que se almacenan en búfer antes de escribir los datos en el destino. Nota Si configura tanto `flushIntervalSeconds` como `flushIntervalEventCount`, la operación de escritura se activa cuando se alcanza cualquiera de los dos umbrales.

Nota

En un trabajo JAR, el código de negocio se carga mediante un cargador de clases distinto al del sistema de registros de Flink. De forma predeterminada, el plugin de registros valida estrictamente los cargadores de clases (isInitStrictly tiene como valor predeterminado true), por lo que dichos registros se omiten y no se exportan a SLS. Para recopilar los registros que imprime el código de negocio en un trabajo JAR, agregue la siguiente configuración al appender de SLS:

<Property name="isInitStrictly">false</Property>

Kafka

Nota

Actualmente no es compatible la autenticación Kerberos para clústeres de Kafka.

Requisitos previos

Realtime Compute for Apache Flink proporciona un plugin de registros KafkaAppender. Dado que Flink carga este plugin mediante un cargador de clases de plugins, debe especificar explícitamente su ruta de paquete para que la aplicación pueda cargarlo correctamente. Puede hacerlo de una de las siguientes maneras:
- Configurar una plantilla de trabajo (afecta a todos los despliegues de este espacio de trabajo)
  
  En la página Configurations de la consola, agregue el siguiente código al campo other configuration.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```
- Configurar un solo despliegue (afecta únicamente al despliegue actual)
  
  En la página Deployments, haga clic en el nombre del despliegue de destino. En la pestaña Configuration y dentro de la sección Parameters, agregue el siguiente código al campo other configuration.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Configuración de registros

<?xml version="1.0" encoding="UTF-8"?>
<Configuration xmlns="http://logging.apache.org/log4j/2.0/config" 
strict="true" packages="com.ververica.platform.logging.appender" status="WARN">  
  <Appenders> 
    <Appender name="StdOut" type="Console"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout"/> 
    </Appender>  
    <Appender name="RollingFile" type="RollingFile" fileName="${sys:log.file}" filePattern="${sys:log.file}.%i"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout"/>  
      <Policies> 
        <SizeBasedTriggeringPolicy size="20 MB"/> 
      </Policies>  
      <DefaultRolloverStrategy max="4"/> 
    </Appender>  
    <Appender type="KafkaVVP" name="KafkaVVPAppender" topic="YOUR-TOPIC-NAME">
        <Layout type="PatternLayout" pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n"/>
        <Property name="bootstrap.servers">YOUR-KAFKA-BOOTSTRAP-SERVERS</Property>
         <Property name="acks">YOUR-ACKS-VALUE</Property>
         <Property name="buffer.memory">YOUR-BUFFER-MEMORY-SIZE</Property>
          <Property name="retries">YOUR-RETRIES-NUMBER</Property>
         <Property name="compression.type">YOUR-COMPRESSION-TYPE</Property>
    </Appender>
    <Appender type="Async" name="AsyncAppender">
        <AppenderRef ref="KafkaVVPAppender"/>
    </Appender>
  </Appenders>
  <Loggers> 
    <Logger level="INFO" name="org.apache.hadoop"/>  
    <Logger level="INFO" name="org.apache.kafka"/>  
    <Logger level="INFO" name="org.apache.zookeeper"/>  
    <Logger level="INFO" name="akka"/>  
    <Logger level="ERROR" name="org.jboss.netty.channel.DefaultChannelPipeline"/>  
    <Logger level="OFF" name="org.apache.flink.runtime.rest.handler.job.JobDetailsHandler"/> 
    {%- for name, level in userConfiguredLoggers -%} 
      <Logger level="{{ level }}" name="{{ name }}"/> 
    {%- endfor -%}
    <Root level="{{ rootLoggerLogLevel }}"> 
      <AppenderRef ref="StdOut"/>
      <AppenderRef ref="RollingFile"/>  
      <AppenderRef ref="AsyncAppender"/> 
    </Root>
  </Loggers>
</Configuration>

Parámetro	Descripción
YOUR-TOPIC-NAME	Nombre del tema de Kafka de destino.
YOUR-KAFKA-BOOTSTRAP-SERVERS	Direcciones de los brokers de Kafka de destino.
YOUR-ACKS-VALUE	Número de réplicas de partición que deben confirmar un mensaje antes de que el productor considere correcta la escritura. Para obtener más información, consulte acks.
YOUR-BUFFER-MEMORY-SIZE	Tamaño del búfer del productor, en bytes.
YOUR-RETRIES-NUMBER	Número de reintentos tras un error de envío.
YOUR-COMPRESSION-TYPE	Tipo de compresión que el productor utiliza para los mensajes. Los valores válidos incluyen none, gzip, snappy, lz4 y zstd.

Haga clic en Save.
Haga clic en Start en la parte superior de la página.

Configurar la exportación de registros para un namespace

Utilice una plantilla para configurar el destino de exportación de registros predeterminado hacia OSS, SLS o Kafka para todos los trabajos dentro de un namespace.

Importante

Una vez que configure esta plantilla, los registros de todos los trabajos nuevos en este namespace se exportarán a OSS, SLS o Kafka.
El cargador de clases del plugin de Flink carga el plugin de registros KafkaAppender que proporciona Realtime Compute for Apache Flink. Antes de usar este plugin, debe especificar su ruta de clase para que la aplicación de Flink lo cargue correctamente. Para ello, en la consola de desarrollo de Realtime Compute for Apache Flink, vaya a la Configurations página y agregue a continuación el código en la sección Other Configuration.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Vaya a la página de configuración de la plantilla de registros del trabajo.
1. Inicie sesión en la consola de Realtime Compute for Apache Flink.
2. En la columna Actions del espacio de trabajo de destino, haga clic en Console. En la consola de desarrollo de Realtime Compute for Apache Flink, seleccione el namespace de destino en la parte superior de la página.
3. En el panel de navegación izquierdo, haga clic en O&M > Configurations.
4. En la pestaña Deployment Defaults, seleccione un tipo de trabajo.
5. En la sección Logging, establezca Logging Profile como Custom Template.
Configure el destino de exportación de registros para todos los trabajos del namespace.

Consulte Configure log export for a single job (XML) para ver el código obligatorio.
Haga clic en save changes.

Configurar salidas de registros por nivel

Con el elemento ThresholdFilter de Log4j2, puede configurar distintas reglas de filtrado para cada appender según el nivel de registro. Esta configuración ofrece varias ventajas:

Flexibilidad: permite establecer distintos niveles de registro para diferentes destinos de almacenamiento externos.
Eficiencia: reduce el procesamiento y la transferencia innecesaria de registros, lo que mejora el rendimiento del sistema.
Claridad: las configuraciones independientes facilitan la comprensión del flujo de registros y simplifican la gestión de niveles.

Siga estos pasos:

En la sección logging, establezca Logging Profile como Custom Template.

Configure la salida de registros.

En el siguiente ejemplo se muestra cómo configurar el sistema para enviar los registros de nivel INFO y superiores a la consola de desarrollo de Realtime Compute for Apache Flink. Al mismo tiempo, solo los registros de nivel ERROR y superiores se envían a Simple Log Service (SLS).

<?xml version="1.0" encoding="UTF-8"?>
<Configuration xmlns="http://logging.apache.org/log4j/2.0/config" strict="true" status="WARN">
  <Appenders>
    <!-- Console Appender configured to output logs of INFO level and higher -->
    <Appender name="StdOut" type="Console">
      <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
      <Layout type="PatternLayout" pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" charset="UTF-8"/>
    </Appender>
    <!-- RollingFile Appender logs messages at the root logger's level (INFO) and higher -->
    <Appender name="RollingFile" type="RollingFile">
      <!-- Configuration remains unchanged -->
      <!-- ... -->
    </Appender>
    <!-- SLS Appender configured to output only ERROR level and above logs -->
    <Appender name="SLS" type="SLS">
      <ThresholdFilter level="ERROR" onMatch="ACCEPT" onMismatch="DENY"/>
      <Layout type="PatternLayout" pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" charset="UTF-8"/>
      <!-- SLS specific properties -->
      <Property name="namespace">YOUR_NAMESPACE</Property>
      <Property name="project">YOUR_SLS_PROJECT</Property>
      <Property name="logStore">YOUR_SLS_LOGSTORE</Property>
      <Property name="endpoint">YOUR_SLS_ENDPOINT</Property>
      <!-- Access credentials and other properties -->
      <!-- ... -->
    </Appender>
    <!-- Other Appenders definitions remain unchanged -->
    <!-- ... -->
  </Appenders>
  <Loggers>
    <!-- Define specific loggers that use the StdOut and SLS appenders -->
    <Logger name="StdOutLogger" level="INFO" additivity="false">
      <AppenderRef ref="StdOut"/>
    </Logger>
    <Logger name="SLSLogger" level="ERROR" additivity="false">
      <AppenderRef ref="SLS"/>
    </Logger>
    <!-- Other Loggers definitions with their specific configurations -->
    <!-- ... -->
    <!-- The Root logger sends output to StdOut and RollingFile, but not SLS, to avoid duplicate logs. -->
    <Root level="INFO">
      <AppenderRef ref="StdOut"/>
      <AppenderRef ref="RollingFile"/>
      <!-- Exclude SLS from Root to prevent duplicate logging in case of other loggers -->
    </Root>
  </Loggers>
</Configuration>

En esta configuración:

Console Appender: utiliza un ThresholdFilter para garantizar que los registros de nivel INFO o superior se exporten a la consola de desarrollo.
El SLS Appender usa un ThresholdFilter para enviar a SLS únicamente los registros de nivel ERROR y superiores. Para más detalles sobre las propiedades del SLS Appender, consulte Configure Output to SLS. Reemplace los marcadores de posición, como YOUR_NAMESPACE y YOUR_SLS_PROJECT, con la información real de su proyecto de SLS.

Nota
Si el tipo del SLS Appender no es SLS, sino un appender personalizado, asegúrese de usar el tipo correcto y de que la clase del appender tenga la lógica necesaria para conectarse a SLS.
StdOutLogger y SLSLogger: envían registros exclusivamente al StdOut Appender y al SLS Appender, respectivamente. Cada uno cuenta con sus propias restricciones de nivel de registro.
Root Logger: este logger se configura con un StdOut Appender y un RollingFile Appender, pero no incluye un SLS Appender. Esto evita enviar registros duplicados a SLS cuando ya existe un logger específico configurado para ese fin.

Para obtener más información sobre los parámetros de configuración de Log4j, consulte Apache Log4j.

Temas relacionados

A job fails to start after you configure logging to SLS
Para consultar los registros de los trabajos, vea View startup and runtime logs.
Si un trabajo no inicia o se ejecuta de forma anómala, consulte View exception logs.
Si los registros de nivel INFO no son suficientes para solucionar problemas, consulte Dynamically change the log level of a running job.
Para consultar los eventos de auditoría de Flink con ActionTrail, vea View Flink audit events.