Realtime Compute for Apache Flink:Konfigurasikan output log pekerjaan

Konfigurasikan output ke 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"
        }
    ]
}

Konfigurasikan output ke OSS

Konfigurasikan output ke Kafka

• Prasyarat

  Plugin logging KafkaAppender untuk komputasi real-time dimuat oleh class loader plugin Flink. Untuk memuat plugin ini secara berhasil dalam aplikasi Flink, Anda harus secara eksplisit menentukan path paketnya. Langkah-langkah berikut menjelaskan cara mengonfigurasi plugin untuk satu pekerjaan. Konfigurasi ini hanya berlaku untuk pekerjaan tersebut.

  Di halaman Job O&M, klik nama pekerjaan target. Pada tab Deployment Details, di bagian Other Configurations dalam area Runtime Parameter Configuration, tambahkan kode berikut.

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

• Antarmuka konfigurasi

  

  Parameter	Deskripsi
  name	Nama kustom untuk appender.
  type	Jenis channel output. Nilainya tetap KafkaVVP. Jangan ubah nilai ini.
  pattern	Format output untuk log. Nilai default adalah %d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n.
  bootstrap.servers	Alamat broker Kafka.
  acks	Menentukan jumlah replika partisi yang harus menerima pesan sebelum produsen menganggap penulisan berhasil. Untuk informasi selengkapnya tentang nilai-nilainya, lihat acks.
  buffer.memory	Ukuran buffer produsen, dalam byte.
  retries	Jumlah kali percobaan ulang mengirim pesan jika gagal.
  compression.type	Jenis kompresi untuk data yang dihasilkan oleh produsen. Nilai yang valid adalah `none`, `gzip`, `snappy`, `lz4`, dan `zstd`.

Konfigurasikan ke 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>

Konfigurasikan ke 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>

Konfigurasikan ke Kafka

• Prasyarat

  Ekstensi log KafkaAppender, yang disediakan oleh komputasi real-time, dimuat melalui class loader plugin Flink. Sebelum menggunakan ekstensi ini, Anda harus secara eksplisit menentukan path paket tempat ekstensi log KafkaAppender berada untuk memastikan aplikasi Flink dapat memuatnya secara berhasil. Untuk melakukan ini, ikuti langkah-langkah berikut:

  • Konfigurasikan templat pekerjaan (berlaku untuk semua pekerjaan dalam proyek ini)

    Pada halaman Configuration Management Konsol pengembangan Flink, tambahkan kode berikut di bawah Other Configurations.

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

  • Konfigurasikan satu pekerjaan (hanya berlaku untuk pekerjaan saat ini)

    Pada halaman Job O&M, klik nama pekerjaan target. Pada tab Deployment Details, tambahkan kode berikut di bagian Runtime Parameter Configuration, di bawah Other Configurations.

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

• Konfigurasi log

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

  Parameter	Deskripsi
  YOUR-TOPIC-NAME	Nama Topik Kafka untuk penulisan.
  YOUR-KAFKA-BOOTSTRAP-SERVERS	Alamat broker Kafka untuk penulisan.
  YOUR-ACKS-VALUE	Menentukan jumlah replika partisi yang harus menerima pesan sebelum produsen menganggap operasi penulisan berhasil. Untuk nilai yang valid, lihat acks.
  YOUR-BUFFER-MEMORY-SIZE	Ukuran buffer produsen, dalam byte.
  YOUR-RETRIES-NUMBER	Jumlah percobaan ulang untuk pengiriman yang gagal.
  YOUR-COMPRESSION-TYPE	Jenis kompresi yang tersedia saat produsen menghasilkan data, termasuk none, gzip, snappy, lz4, atau zstd.

  Catatan

  Anda juga dapat mengonfigurasi semua parameter yang didukung oleh klien Apache Kafka. Untuk informasi selengkapnya, lihat Apache Kafka.