Realtime Compute for Apache Flink:ログエクスポートの設定

SLS へのログのエクスポート

OSS へのログのエクスポート

Kafka へのログのエクスポート

• 前提条件

  以下の手順を実行して、下のコードを追加します:

  1. ジョブ詳細ページで、[設定] タブを選択します。

  2. [パラメーター] セクションで、[編集] をクリックします。

  3. 次のコードをコピーして [その他の設定] フィールドに貼り付けます。

  plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender

  このコードスニペットは、Realtime Compute for Apache Flink が提供するロギングプラグインである KafkaAppender パッケージのパスを明示的に指定します。これにより、Realtime Compute for Apache Flink がロギングプラグインをロードできるようになります。

• ログエクスポートの設定

  

  項目	説明
  name	カスタム Appender 名を入力します。
  type	出力チャンネルタイプ。KafkaVVP に固定されています。
  pattern	ログ出力形式。 デフォルト値は %d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n です。
  bootstrap.servers	Kafka ブローカーのエンドポイントを入力します。
  acks	Kafka プロデューサーがメッセージがトピックに書き込まれたと判断する前に、メッセージを受信したパーティションレプリカの数を指定します。詳細については、「acks」をご参照ください。
  buffer.memory	プロデューサーのバッファーサイズを入力します。単位: バイト。
  retries	メッセージの送信に失敗した後に許可される最大再試行回数を入力します。
  compression.type	プロデューサーがデータを生成するために使用する圧縮タイプを入力します。有効な値: none, gzip, snappy, lz4, zstd。

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"/>  
      
      <!-- 最終的に有効なログパスは ${baseUri}/logs/${namespace}/${deploymentId}/{jobId}/ です -->
      <Property name="namespace">{{ namespace }}</Property> <!-- この行は変更しないでください -->
      <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>

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"/>  

      <!-- 最終的に有効なログパスは ${baseUri}/logs/${namespace}/${deploymentId}/{jobId}/ です -->
      <Property name="namespace">{{ namespace }}</Property> <!-- この行は変更しないでください -->
      <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>

Kafka へのログのエクスポート

• 前提条件

  Realtime Compute for Apache Flink が提供するロギングプラグインである KafkaAppender パッケージのパスを明示的に指定します。これにより、Realtime Compute for Apache Flink がロギングプラグインをロードできるようになります。次の手順を実行します:

  • 名前空間内のすべてのジョブに設定を適用します:

    1. O&M > 設定 に移動します。

    2. [その他の設定] セクションで、次のコードをコピーして貼り付け、変更を保存します。

  plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender

  • 単一のジョブに設定を適用します:

    1. デプロイメント詳細ページで、[設定] タブを選択します。

    2. [パラメーター] セクションで、[編集] をクリックします。

    3. 次のコードを [その他の設定] フィールドに追加し、変更を保存します。

    plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender

• ログエクスポートの設定

  <?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>

  プレースホルダー	説明
  YOUR-TOPIC-NAME	対象の Kafka トピックの名前に置き換えます。
  YOUR-KAFKA-BOOTSTRAP-SERVERS	Kafka ブローカーのエンドポイントに置き換えます。
  YOUR-ACKS-VALUE	プロデューサーがメッセージがトピックに書き込まれたと判断する前にメッセージを受信する必要があるパーティションレプリカの数に置き換えます。詳細については、「acks」をご参照ください。
  YOUR-BUFFER-MEMORY-SIZE	プロデューサーバッファーのサイズに置き換えます。単位: バイト。
  YOUR-RETRIES-NUMBER	メッセージの送信に失敗した後に許可される最大再試行回数に置き換えます。
  YOUR-COMPRESSION-TYPE	プロデューサーがデータを生成するために使用できる圧縮タイプに置き換えます。有効な値: none, gzip, snappy, lz4, zstd。

  説明

  Apache Kafka クライアントでサポートされているパラメーターを設定することもできます。詳細については、「Apache Kafka」をご参照ください。