Realtime Compute for Apache Flink:ジョブログの出力設定

SLS 向けの設定

{

    "Version": "1",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "log:Get*",
                "log:PostLogStoreLogs"
            ],
            "Resource": "*"
        }
    ]
}

{
    "Version": "1",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "log:PostLogStoreLogs",
                "log:GetLogStore"
            ],
            "Resource": "acs:log:cn-beijing:152940222687****:project/test-vvp-sls/logstore/test-ltest"
        }
    ]
}

OSS 向けの設定

Kafka 向けの設定

• 前提条件

  Realtime Compute for Apache Flink が提供する KafkaAppender ログプラグインは、Flink プラグインクラスローダーによって読み込まれます。このプラグインを使用する前に、Flink アプリケーションがプラグインを正常に読み込めるよう、KafkaAppender ログプラグインのパッケージパスを明示的に指定する必要があります。単一ジョブ向け（現在のジョブにのみ適用）のプラグイン設定を行うには、以下の手順を実行します：

  「ジョブ O&M」ページで対象のジョブ名をクリックします。「デプロイメントの詳細」タブの「実行パラメーター設定」セクション内、「その他の設定」に次のコードを追加します。

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

• 設定インターフェイス

  

  パラメーター	説明
  name	Appender のカスタム名です。
  type	出力チャネルの種類です。値は固定で KafkaVVP となります。変更する必要はありません。
  pattern	ログの出力フォーマットです。 デフォルト値は %d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n です。
  bootstrap.servers	ログを書き込む Kafka ブローカーのアドレスです。
  acks	メッセージの書き込みが成功したと見なすために、パーティションレプリカが受信しなければならない数です。詳細については、「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 ログプラグインは、Flink プラグインクラスローダーによって読み込まれます。このプラグインを使用する前に、Flink アプリケーションがプラグインを正常に読み込めるよう、KafkaAppender ログプラグインのパッケージパスを明示的に指定する必要があります。以下のいずれかの方法を使用できます：

  • ジョブテンプレートの設定（プロジェクト内のすべてのジョブに有効）

    Flink 開発コンソールの「構成管理」ページの「その他の設定」セクションに、次のコードを追加します。

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

  • 単一ジョブの設定（現在のジョブにのみ有効）

    「ジョブ O&M」ページで対象のジョブ名をクリックします。「デプロイメントの詳細」タブの「実行パラメーター設定」エリア内の「追加設定」セクションに、次のコードを追加します。

    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 Topic の名前です。
  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」をご参照ください。