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

数値パラメータのデフォルト値は次のとおりです：flushIntervalSeconds は 10、flushIntervalEventCount は 100、rollingBytes は 1048576 です。

Kafka

• 前提条件

  KafkaAppender プラグインは Flink プラグインクラスローダーによってロードされるため、Flink 設定でそのパッケージパスを明示的に宣言する必要があります。これにより、アプリケーションが現在のデプロイメント用のプラグインをロードできるようになります。

  [デプロイ] ページで、お使いのデプロイの名前をクリックします。[設定] タブの [追加の Flink 設定] セクションで、[その他の設定] フィールドに次のパラメーターを追加します。

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

• 設定

  パラメータ	説明
  name	アペンダーのカスタム名。
  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"/>  
      <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 はこのプラグインをプラグインクラスローダーを使用して読み込むため、アプリケーションが正常に読み込めるようにそのパッケージパスを明示的に指定する必要があります。次のいずれかの方法で実行できます：

  • ジョブテンプレートの設定 (このワークスペース内のすべてのデプロイメントに影響)：

    コンソールの[設定]ページで、[その他の設定]フィールドに次のコードを追加します。

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

  • 単一デプロイメントの設定 (現在のデプロイメントのみに影響)：

    [デプロイ] ページで、対象のデプロイの名前をクリックします。[デプロイ詳細] タブの [パラメーター] セクションで、[その他の設定] フィールドに次のコードを追加します。

    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 です。