Configure Flink job log export - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Observações

Após configurar a exportação de logs para OSS, SLS ou Kafka, reinicie o job.
Se você configurar a exportação de logs para outro serviço de armazenamento, mas não desativar o recurso de arquivamento de logs, eles continuarão sendo salvos no OSS ou no armazenamento totalmente gerenciado configurado para seu workspace. Caso desative o arquivamento de logs, não será mais possível visualizar os logs do job no console do Flink.

Na página deployment details, na seção log configuration, use o alternador na linha log archiving para ativar ou desativar esse recurso.
Use ${secret_values.xxxx} na configuração de log para referenciar variáveis armazenadas no gerenciamento de variáveis. Para obter mais informações, consulte project variable.

Configurar exportação de logs de um único job

Console

Acesse a página de configuração de log de uma implantação.
1. Faça login no console do Realtime Compute for Apache Flink.
2. Na coluna Actions do workspace desejado, clique em Console.
3. No painel de navegação à esquerda, clique em O&M > Deployments. Em seguida, clique no nome da implantação desejada.
4. Na aba Configuration, na seção Logging, clique em Edit à direita.
5. Para Logging Profile, selecione Custom Template.
Clique em Apply with System Profiles > default.
Clique em Add Appender e selecione um destino.

Configure os parâmetros do appender selecionado.

Para rotear logs de diferentes níveis para destinos distintos, configure regras de filtragem para o appender. Para obter mais informações, consulte Configure log filtering based on log levels.

SLS

Parâmetro	Descrição
name	Nome personalizado para o appender.
type	Tipo do appender. Este valor é fixo como `SLS`. Não o altere.
pattern	Formato de saída dos logs. O valor padrão é `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`, que gera entradas de log como `2023-10-01 14:23:45,678{GMT+8} INFO com.example.MyClass - This is a test log message`.
flushIntervalSeconds	Intervalo, em segundos, para liberação dos dados de log no destino.
flushIntervalEventCount	Número de entradas de log a serem agrupadas antes da gravação no destino de armazenamento. Nota Se tanto `flushIntervalSeconds` quanto `flushIntervalEventCount` estiverem configurados, a operação de gravação será acionada quando qualquer um dos limiares for atingido, o que ocorrer primeiro.
authenticationMode	AccessKey SLS Token Nota Ao selecionar SLS Token, a exportação de logs só pode ser feita para um Logstore em um Project que esteja na mesma região do seu workspace do Flink. Esse recurso é suportado apenas no VVR 11.5 e versões posteriores.
project	Nome do Project do SLS.
logStore	Nome do Logstore.
endpoint	Endpoint privado da região do serviço SLS. Para obter mais informações, consulte Endpoints.
accessKeyId	AccessKey ID e AccessKey Secret da conta usada para o serviço SLS. Para saber como obter um par de AccessKey, consulte Obtain an AccessKey pair. Para evitar riscos de segurança associados ao uso de chaves em texto simples, recomendamos armazenar os valores de AccessKey em variáveis de namespace. Para obter mais informações, consulte Namespace variables. Nota Se seus serviços SLS e Realtime Compute for Apache Flink estiverem em contas Alibaba Cloud diferentes, conceda permissão à conta do Flink para gravar no SLS. Para detalhes, consulte Create a custom RAM policy. Veja abaixo exemplos de políticas: Escopo irrestrito do SLS `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:Get", "log:PostLogStoreLogs" ], "Resource": "" } ] }` Escopo específico de recurso do 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

Os valores padrão para os parâmetros numéricos são: flushIntervalSeconds é 10, flushIntervalEventCount é 100 e rollingBytes é 1048576.

Parâmetro	Descrição
name	Nome personalizado para o appender.
type	Tipo do appender. Este valor é fixo como `OSS`. Não o altere.
pattern	Formato de saída dos logs. O valor padrão é `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`.
baseUri	Nome do bucket do OSS.
endpoint	Endpoint da região do seu serviço OSS. Para obter mais informações, consulte Regions and endpoints. Utilize o endpoint da coluna Endpoint (region) correspondente à linha Access from ECS over the VPC (internal network).
accessKeyId	AccessKey ID e AccessKey Secret da conta usada para o serviço OSS. Para saber como obter um par de AccessKey, consulte Obtain an AccessKey pair. Para evitar riscos de segurança associados ao uso de chaves em texto simples, recomendamos armazenar os valores de AccessKey em variáveis de namespace. Para obter mais informações, consulte Namespace variables. Nota Esses parâmetros são necessários apenas se o bucket do OSS de destino estiver em uma conta Alibaba Cloud diferente daquela do seu serviço Flink. Se estiverem na mesma conta, remova esses parâmetros da configuração.
accessKeySecret
flushIntervalSeconds	Intervalo, em segundos, para liberação dos dados de log no destino.
flushIntervalEventCount	Número de entradas de log a serem agrupadas antes da gravação no destino de armazenamento. Nota Se tanto `flushIntervalSeconds` quanto `flushIntervalEventCount` estiverem configurados, a operação de gravação será acionada quando qualquer um dos limiares for atingido, o que ocorrer primeiro.
rollingBytes	Tamanho máximo de um único arquivo de log no OSS. Ao atingir esse tamanho, os dados subsequentes serão gravados em um novo arquivo de log.

Kafka

Nota

Clusters Kafka com autenticação Kerberos ativada não são suportados.

Pré-requisitos

Como o plug-in KafkaAppender é carregado pelo class loader de plug-ins do Flink, declare explicitamente seu caminho de pacote na configuração do Flink. Isso permite que a aplicação carregue o plug-in para a implantação atual.

Na página Deployments, clique no nome da sua implantação. Na aba Configuration, na seção Parameters, adicione o seguinte parâmetro ao campo Other Configuration.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Configuração

Parâmetro	Descrição
name	Nome personalizado para o appender.
type	Tipo do appender. Este valor é fixo como `KafkaVVP`. Não o altere.
pattern	Formato de saída dos logs. O valor padrão é `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`.
bootstrap.servers	Endereços dos brokers do Kafka.
acks	Número de réplicas de partição que devem receber uma mensagem antes que o producer considere a gravação bem-sucedida. Para obter mais informações, consulte acks.
buffer.memory	Memória total, em bytes, que o producer pode usar para armazenar registros em buffer.
retries	Número de vezes que o producer tentará reenviar uma mensagem após uma falha transitória.
compression.type	Tipo de compactação para dados gerados pelo producer. Valores válidos: `none`, `gzip`, `snappy`, `lz4` e `zstd`.

Clique em Save.
Na parte superior da página, clique em Start.

Método XML

Para configurar o logging de uma implantação, acesse sua página de configuração de log.
1. Faça login no console do Realtime Compute for Apache Flink.
2. Na coluna Actions do workspace desejado, clique em Console.
3. No painel de navegação à esquerda, clique em O&M > Deployments e, em seguida, clique no nome da implantação desejada.
4. Na aba Configuration, na seção Logging, clique em Edit.
5. Defina Logging Profile como Custom Template.

Configure a saída de log.

Dependendo do seu destino de armazenamento, copie a configuração correspondente, cole-a na caixa de entrada e atualize os parâmetros com os detalhes do seu armazenamento. Para rotear logs de diferentes níveis para destinos distintos, configure regras de filtragem para cada Appender. Para obter mais informações, 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	Descrição
YOUR-BUCKET-NAME	Substitua pelo nome do seu bucket do OSS.
YOUR-ENDPOINT	Substitua pelo seu endpoint do OSS. Para obter mais informações, consulte Regions and endpoints. Este é o valor da coluna Endpoint na linha Access from ECS over the VPC (internal network).
YOUR-OSS-ACCESSKEYID	Use o AccessKey ID e o AccessKey Secret da sua conta do OSS. Para obter mais informações sobre como obtê-los, consulte Obtain an AccessKey pair. Para evitar a exposição de credenciais em texto simples, este exemplo usa variáveis de namespace para especificar os valores de AccessKey. Para obter mais informações, consulte Namespace variables. Nota Esses parâmetros são necessários apenas se o bucket do OSS e seu serviço Realtime Compute for Apache Flink estiverem em contas diferentes. Se o bucket estiver na mesma conta, omita esses parâmetros.
YOUR-OSS-ACCESSKEYSECRET
flushIntervalSeconds	Intervalo, em segundos, para gravação dos dados de log no destino de armazenamento.
flushIntervalEventCount	Número de entradas de log a serem armazenadas em buffer antes da gravação dos dados no destino. Nota Se tanto `flushIntervalSeconds` quanto `flushIntervalEventCount` estiverem configurados, a operação de gravação será acionada quando qualquer um dos limiares for atingido.
rollingBytes	Tamanho máximo de um único arquivo de log no OSS. Quando um arquivo atinge esse tamanho, um novo arquivo de log é criado para os logs subsequentes.

Nota

Em um job JAR, o código de negócio é carregado por um class loader separado, diferente do class loader do sistema de logging do Flink. Por padrão, o plug-in de logging valida estritamente os class loaders (isInitStrictly tem como padrão true), portanto, tais logs são ignorados e não exportados para o OSS. Para coletar os logs impressos pelo código de negócio em um job JAR, adicione a seguinte configuração ao Appender do 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

As propriedades namespace, deploymentId, jobId e deploymentName no código são variáveis Twig. Não as modifique. Caso contrário, o job falhará ao iniciar.

Parâmetro	Descrição
YOUR-SLS-PROJECT	Substitua pelo nome do seu Project do SLS.
YOUR-SLS-LOGSTORE	Substitua pelo nome do seu Logstore do SLS.
YOUR-SLS-ENDPOINT	Substitua pelo endpoint privado da região do seu serviço SLS. Para obter mais informações, consulte Service endpoints.
YOUR-SLS-ACCESSKEYID	Use o AccessKey ID e o AccessKey Secret da sua conta do SLS. Para obter mais informações sobre como obtê-los, consulte Obtain an AccessKey pair. Para evitar a exposição de credenciais em texto simples, este exemplo usa variáveis de namespace para especificar os valores de AccessKey. Para obter mais informações, consulte Namespace variables. Nota Se seus serviços SLS e Realtime Compute for Apache Flink estiverem em contas diferentes, conceda permissões de gravação no SLS à conta do Flink. Para obter mais informações, consulte Create a custom RAM policy. Use uma das seguintes políticas: Escopo irrestrito do SLS `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:Get", "log:PostLogStoreLogs" ], "Resource": "" } ] }` Com escopo para um recurso específico do SLS. Exemplo: `{ "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, em segundos, para gravação dos dados de log no destino de armazenamento.
flushIntervalEventCount	Número de entradas de log a serem armazenadas em buffer antes da gravação dos dados no destino. Nota Se tanto `flushIntervalSeconds` quanto `flushIntervalEventCount` estiverem configurados, a operação de gravação será acionada quando qualquer um dos limiares for atingido.

Nota

Em um job JAR, o código de negócio é carregado por um class loader separado, diferente do class loader do sistema de logging do Flink. Por padrão, o plug-in de logging valida estritamente os class loaders (isInitStrictly tem como padrão true), portanto, tais logs são ignorados e não exportados para o SLS. Para coletar os logs impressos pelo código de negócio em um job JAR, adicione a seguinte configuração ao Appender do SLS:

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

Kafka

Nota

A autenticação Kerberos para clusters Kafka não é suportada atualmente.

Pré-requisitos

O Realtime Compute for Apache Flink fornece um plug-in de logging KafkaAppender. Como o Flink carrega esse plug-in usando um class loader de plug-ins, especifique explicitamente seu caminho de pacote para que sua aplicação possa carregá-lo com sucesso. Faça isso de uma das seguintes maneiras:
- Configurar um modelo de job (afeta todas as implantações neste workspace)
  
  Na página Configurations do console, adicione o seguinte código ao campo other configuration.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```
- Configurar uma única implantação (afeta apenas a implantação atual)
  
  Na página Deployments, clique no nome da implantação desejada. Na aba Configuration, na seção Parameters, adicione o seguinte código ao campo other configuration.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Configuração de logging

<?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	Descrição
YOUR-TOPIC-NAME	Nome do tópico Kafka de destino.
YOUR-KAFKA-BOOTSTRAP-SERVERS	Endereços dos brokers Kafka de destino.
YOUR-ACKS-VALUE	Número de réplicas de partição que devem confirmar uma mensagem antes que o producer considere a gravação bem-sucedida. Para obter mais informações, consulte acks.
YOUR-BUFFER-MEMORY-SIZE	Tamanho do buffer do producer, em bytes.
YOUR-RETRIES-NUMBER	Número de tentativas após uma falha de envio.
YOUR-COMPRESSION-TYPE	Tipo de compactação que o producer usa para mensagens. Os valores válidos incluem none, gzip, snappy, lz4 e zstd.

Clique em Save.
Clique em Start na parte superior da página.

Configurar exportação de logs para um namespace

Use um modelo para configurar o destino padrão de exportação de logs para OSS, SLS ou Kafka para todos os jobs dentro de um namespace.

Importante

Após configurar este modelo, os logs de todos os novos jobs neste namespace serão exportados para OSS, SLS ou Kafka.
O class loader de plug-ins do Flink carrega o plug-in de log KafkaAppender fornecido pelo Realtime Compute for Apache Flink. Antes de usar esse plug-in, especifique seu caminho de classe para permitir que sua aplicação Flink o carregue. Para fazer isso, no console de desenvolvimento do Realtime Compute for Apache Flink, acesse a página Configurations e adicione o seguinte código à seção Other Configuration.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Acesse a página de configuração do modelo de log de job.
1. Faça login no console do Realtime Compute for Apache Flink.
2. Na coluna Actions do workspace desejado, clique em Console. No console de desenvolvimento do Realtime Compute for Apache Flink, selecione o namespace desejado na parte superior da página.
3. No painel de navegação à esquerda, clique em O&M > Configurations.
4. Na aba Deployment Defaults, selecione um tipo de job.
5. Na seção Logging, defina Logging Profile como Custom Template.
Configure o destino de exportação de logs para todos os jobs no namespace.

Para o código necessário, consulte Configure log export for a single job (XML).
Clique em save changes.

Configurar saídas de log baseadas em nível

Utilize o ThresholdFilter do Log4j2 para configurar diferentes regras de filtragem para cada appender com base no nível de log. Essa configuração oferece várias vantagens:

Flexibilidade: Permite definir diferentes níveis de log para diferentes destinos de armazenamento externo.
Eficiência: Reduz o processamento e a transferência desnecessários de logs, melhorando o desempenho do sistema.
Clareza: Configurações separadas tornam o fluxo de logs mais claro e simplificam o gerenciamento de níveis.

Siga estas etapas:

Na seção logging, defina Logging Profile como Custom Template.

Configure a saída de log.

O exemplo a seguir mostra como configurar o sistema para enviar logs de nível INFO e superiores para o console de desenvolvimento do Realtime Compute for Apache Flink, enquanto envia apenas logs de nível ERROR e superiores para o 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>

Nesta configuração:

Console Appender: Usa um ThresholdFilter para garantir que logs de nível INFO ou superior sejam exportados para o console de desenvolvimento.
O SLS Appender usa um ThresholdFilter para enviar apenas logs de nível ERROR e superior para o SLS. Para detalhes sobre as propriedades do SLS Appender, consulte Configure Output to SLS. Substitua placeholders como YOUR_NAMESPACE e YOUR_SLS_PROJECT pelas informações reais do seu projeto SLS.

Nota
Se o tipo do SLS Appender não for SLS, mas um appender personalizado, certifique-se de usar o tipo correto e de que a classe do appender possua a lógica necessária para se conectar ao SLS.
StdOutLogger e SLSLogger: Enviam logs apenas para o StdOut Appender e o SLS Appender, respectivamente, e cada um possui suas próprias restrições de nível de log.
Root Logger: Este logger é configurado com um StdOut Appender e um RollingFile Appender, mas não inclui um SLS Appender. Isso evita o envio de logs duplicados para o SLS caso um logger específico já esteja configurado para fazê-lo.

Para obter mais informações sobre os parâmetros de configuração do Log4j, consulte Apache Log4j.

Tópicos relacionados

A job fails to start after you configure logging to SLS
Para visualizar logs de jobs, consulte View startup and runtime logs.
Se um job falhar ao iniciar ou apresentar execução anormal, consulte View exception logs.
Se os logs de nível INFO forem insuficientes para solução de problemas, consulte Dynamically change the log level of a running job.
Para visualizar eventos de auditoria do Flink com o ActionTrail, consulte View Flink audit events.