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 設定でそのパッケージパスを明示的に宣言する必要があります。これにより、アプリケーションが現在のデプロイメント用のプラグインを読み込むことができます。

  [Deployments] ページで、デプロイメントの名前をクリックします。[Configuration] タブの [Additional Flink Configuration] セクションで、[Other Configuration] フィールドに次のパラメータを追加します。

  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>

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

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

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