Tüm Ürünler
Search
Belge Merkezi

Realtime Compute for Apache Flink:İş günlüğü dışa aktarımını yapılandırın

Son Güncelleme:Oct 04, 2026

Flink konsolundaki iş içgörüsü sayfasında iş günlüklerini görüntüleyebilir veya bunları OSS, SLS ya da Kafka gibi harici bir depolama hizmetine aktarıp çıkış için günlük düzeyini yapılandırabilirsiniz. Bu konuda bu işlemin nasıl yapılacağı açıklanmaktadır.

Notlar

  • Günlük dışa aktarımını OSS, SLS veya Kafka'ya yapılandırdıktan sonra işi yeniden başlatmanız gerekir.

  • Günlük dışa aktarımını başka bir depolama hizmetine yapılandırır ancak günlük arşivleme özelliğini devre dışı bırakmazsanız, günlükler çalışma alanınız için yapılandırılan OSS veya tamamen yönetilen depolamaya kaydedilmeye devam eder. Günlük arşivlemeyi devre dışı bırakırsanız Flink konsolunda artık iş günlüklerini görüntüleyemezsiniz.

    deployment details sayfasında, log configuration bölümünde, log archiving satırındaki geçiş anahtarını kullanarak bu özelliği etkinleştirin veya devre dışı bırakın.

  • Günlük yapılandırmasında değişken yönetiminde saklanan değişkenlere başvurmak için ${secret_values.xxxx} ifadesini kullanabilirsiniz. Daha fazla bilgi için bkz. proje değişkeni.

Tek bir iş için günlük dışa aktarımını yapılandırın

Console

  1. Bir dağıtımın günlük yapılandırma sayfasına gidin.

    1. Realtime Compute for Apache Flink konsolunda oturum açın.

    2. Hedef çalışma alanının Actions sütununda Console seçeneğini tıklayın.

    3. Sol gezinme bölmesinde O&M > Deployments seçeneklerine gidin. Ardından hedef dağıtımın adını tıklayın.

    4. Configuration sekmesinde, Logging bölümünün sağ tarafındaki Edit seçeneğini tıklayın.

    5. Logging Profile için Custom Template seçeneğini belirleyin.

  2. Apply with System Profiles > default seçeneklerini tıklayın.

  3. Add Appender seçeneğini tıklayın ve bir hedef konum belirleyin.

  4. Seçilen ekleyici (appender) için parametreleri yapılandırın.

    Farklı düzeylerdeki günlükleri farklı hedeflere yönlendirmek için ekleyiciye filtreleme kuralları ekleyin. Daha fazla bilgi için bkz. Günlük düzeylerine göre günlük filtrelemeyi yapılandırma.

    SLS

    Parametre

    Açıklama

    name

    Ekleyici için özel bir ad.

    type

    Ekleyici türü. Bu değer SLS olarak sabittir. Değeri değiştirmeyin.

    pattern

    Günlüklerin çıkış biçimi.

    Varsayılan değer %d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n şeklindedir; bu değer 2023-10-01 14:23:45,678{GMT+8} INFO com.example.MyClass - This is a test log message gibi günlük girişleri oluşturur.

    flushIntervalSeconds

    Günlük verilerinin hedefe yazılma aralığı (saniye cinsinden).

    flushIntervalEventCount

    Depolama hedefine yazılmadan önce toplu işlenecek günlük girişi sayısı.

    Not

    Hem flushIntervalSeconds hem de flushIntervalEventCount yapılandırılırsa yazma işlemi, hangi eşik önce karşılanırsa o anda tetiklenir.

    authenticationMode

    • AccessKey

    • SLS Token

      Not

      SLS Token seçeneğini belirlerseniz günlükleri yalnızca Flink çalışma alanınızla aynı bölgedeki bir Project içindeki Logstore'a aktarabilirsiniz. Bu özellik yalnızca VVR 11.5 ve sonraki sürümlerde desteklenir.

    project

    SLS Project adı.

    logStore

    Logstore adı.

    endpoint

    SLS hizmet bölgesi için özel uç nokta. Daha fazla bilgi için bkz. Uç noktalar.

    accessKeyId

    SLS hizmeti için kullanılan hesabın AccessKey ID ve AccessKey Secret değerleri. AccessKey çiftinin nasıl alınacağı hakkında bilgi edinmek için bkz. AccessKey çifti alma.

    Düz metin anahtarları kullanmanın güvenlik risklerinden kaçınmak için AccessKey değerlerini ad alanı değişkenlerinde saklamanızı öneririz. Daha fazla bilgi için bkz. Ad alanı değişkenleri.

    Not

    SLS ve Realtime Compute for Apache Flink hizmetleriniz farklı Alibaba Cloud hesaplarında bulunuyorsa Flink hesabına SLS'ye yazma izni vermeniz gerekir. Ayrıntılar için bkz. Özel RAM ilkesi oluşturma. Aşağıda ilke örnekleri verilmiştir:

    Kısıtlamasız SLS kapsamı

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

    Belirli SLS kaynak kapsamı

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

    accessKeySecret

    OSS

    Sayısal parametrelerin varsayılan değerleri şöyledir: flushIntervalSeconds 10, flushIntervalEventCount 100 ve rollingBytes 1048576.

    Parametre

    Açıklama

    name

    Ekleyici için özel bir ad.

    type

    Ekleyici türü. Bu değer OSS olarak sabittir. Değeri değiştirmeyin.

    pattern

    Günlüklerin çıkış biçimi.

    Varsayılan değer %d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n şeklindedir.

    baseUri

    OSS paket adı.

    endpoint

    OSS hizmetinizin bulunduğu bölgenin uç noktası. Daha fazla bilgi için bkz. Bölgeler ve uç noktalar.

    Endpoint (region) sütunundaki ve Access from ECS over the VPC (internal network) satırına karşılık gelen uç noktayı kullanın.

    accessKeyId

    OSS hizmeti için kullanılan hesabın AccessKey ID ve AccessKey Secret değerleri. AccessKey çiftinin nasıl alınacağı hakkında bilgi edinmek için bkz. AccessKey çifti alma.

    Düz metin anahtarları kullanmanın güvenlik risklerinden kaçınmak için AccessKey değerlerini ad alanı değişkenlerinde saklamanızı öneririz. Daha fazla bilgi için bkz. Ad alanı değişkenleri.

    Not

    Bu parametreler yalnızca hedef OSS paketi Flink hizmetinizden farklı bir Alibaba Cloud hesabındaysa gereklidir. Aynı hesaptalarsa bu parametreleri yapılandırmadan kaldırın.

    accessKeySecret

    flushIntervalSeconds

    Günlük verilerinin hedefe yazılma aralığı (saniye cinsinden).

    flushIntervalEventCount

    Depolama hedefine yazılmadan önce toplu işlenecek günlük girişi sayısı.

    Not

    Hem flushIntervalSeconds hem de flushIntervalEventCount yapılandırılırsa yazma işlemi, hangi eşik önce karşılanırsa o anda tetiklenir.

    rollingBytes

    OSS'deki tek bir günlük dosyasının maksimum boyutu. Bu boyuta ulaşıldığında sonraki veriler yeni bir günlük dosyasına yazılır.

    Kafka

    Not

    Kerberos kimlik doğrulaması etkinleştirilmiş Kafka kümeleri desteklenmez.

    • Önkoşullar

      KafkaAppender eklentisi Flink eklenti sınıf yükleyicisi tarafından yüklendiğinden uygulamanın geçerli dağıtım için eklentiyi yükleyebilmesi amacıyla Flink yapılandırmasında paket yolunu açıkça bildirmeniz gerekir.

      Deployments sayfasında dağıtımınızın adını tıklayın. Configuration sekmesinde, Parameters bölümündeki Other Configuration alanına aşağıdaki parametreyi ekleyin.

      plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
    • Yapılandırma


      Parametre


      Açıklama


      name


      Ekleyici için özel bir ad.


      type


      Ekleyici türü. Bu değer KafkaVVP olarak sabittir. Değeri değiştirmeyin.


      pattern

      Günlüklerin çıkış biçimi.

      Varsayılan değer %d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n şeklindedir.


      bootstrap.servers


      Kafka broker adresleri.


      acks


      Üreticinin yazma işlemini başarılı olarak kabul etmesi için mesajı alması gereken bölüm çoğaltma sayısı. Daha fazla bilgi için bkz. acks.


      buffer.memory


      Üreticinin kayıtları arabelleğe almak için kullanabileceği toplam bellek (bayt cinsinden).


      retries


      Geçici bir hata oluştuğunda üreticinin mesajı yeniden göndermeyi deneme sayısı.


      compression.type


      Üretici tarafından oluşturulan verilerin sıkıştırma türü. Geçerli değerler: none, gzip, snappy, lz4 ve zstd.

  5. Save seçeneğini tıklayın.

  6. Sayfanın üst kısmında Start seçeneğini tıklayın.

XML yöntemi

  1. Bir dağıtım için günlük kaydını yapılandırmak üzere günlük yapılandırma sayfasına gidin.

    1. Realtime Compute for Apache Flink konsolunda oturum açın.

    2. Hedef çalışma alanının Actions sütununda Console seçeneğini tıklayın.

    3. Sol gezinme bölmesinde O&M > Deployments seçeneklerine gidin ve ardından hedef dağıtımın adını tıklayın.

    4. Configuration sekmesinde, Logging bölümündeki Edit seçeneğini tıklayın.

    5. Logging Profile değerini Custom Template olarak ayarlayın.

  2. Günlük çıkışını yapılandırın.

    Depolama hedefinize bağlı olarak ilgili yapılandırmayı kopyalayın, giriş kutusuna yapıştırın ve parametreleri depolama ayrıntılarınızla güncelleyin. Farklı düzeylerdeki günlükleri farklı hedeflere yönlendirmek için her Appender için filtreleme kuralları yapılandırabilirsiniz. Daha fazla bilgi için bkz. Farklı günlük düzeyleri için günlük çıkışlarını yapılandırma.

    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>

    Parametre

    Açıklama

    YOUR-BUCKET-NAME

    Bu değeri OSS paket adınızla değiştirin.

    YOUR-ENDPOINT

    Bu değeri OSS uç noktanızla değiştirin. Daha fazla bilgi için bkz. Bölgeler ve uç noktalar.

    Bu değer, Access from ECS over the VPC (internal network) satırındaki Endpoint sütunundan alınır.

    YOUR-OSS-ACCESSKEYID

    OSS hesabınız için AccessKey ID ve AccessKey secret değerlerini kullanın. Bunların nasıl alınacağı hakkında daha fazla bilgi için bkz. AccessKey çifti alma.

    Kimlik bilgilerinin düz metin olarak açığa çıkmasını önlemek için bu örnekte AccessKey değerlerini belirtmek üzere ad alanı değişkenleri kullanılır. Daha fazla bilgi için bkz. Ad alanı değişkenleri.

    Not

    Bu parametreler yalnızca OSS paketi ile Realtime Compute for Apache Flink hizmetiniz farklı hesaplardaysa gereklidir. Paket aynı hesaptaysa bu parametreleri atlayabilirsiniz.

    YOUR-OSS-ACCESSKEYSECRET

    flushIntervalSeconds

    Günlük verilerinin depolama hedefine yazılma aralığı (saniye cinsinden).

    flushIntervalEventCount

    Veriler hedefe yazılmadan önce arabelleğe alınacak günlük girişi sayısı.

    Not

    Hem flushIntervalSeconds hem de flushIntervalEventCount yapılandırılırsa yazma işlemi, hangi eşik karşılanırsa o anda tetiklenir.

    rollingBytes

    OSS'deki tek bir günlük dosyasının maksimum boyutu. Bir dosya bu boyuta ulaştığında sonraki günlükler için yeni bir günlük dosyası oluşturulur.

    Not

    JAR işinde iş kodu, Flink günlük sistemi sınıf yükleyicisinden farklı olan ayrı bir sınıf yükleyici tarafından yüklenir. Varsayılan olarak günlük eklentisi sınıf yükleyicilerini katı bir şekilde doğrular (isInitStrictly varsayılan olarak true değerine sahiptir); bu nedenle bu tür günlükler atlanır ve OSS'e aktarılmaz. JAR işindeki iş kodu tarafından yazdırılan günlükleri toplamak için OSS Appender'a aşağıdaki yapılandırmayı ekleyin:

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

    Kodda yer alan namespace, deploymentId, jobId ve deploymentName özellikleri Twig değişkenleridir. Bunları değiştirmeyin. Aksi takdirde iş başlamaz.

    Parametre

    Açıklama

    YOUR-SLS-PROJECT

    Bu değeri SLS Project adınızla değiştirin.

    YOUR-SLS-LOGSTORE

    Bu değeri SLS Logstore adınızla değiştirin.

    YOUR-SLS-ENDPOINT

    Bu değeri SLS hizmetinizin bölgesi için özel uç nokta ile değiştirin. Daha fazla bilgi için bkz. Hizmet uç noktaları.

    YOUR-SLS-ACCESSKEYID

    SLS hesabınız için AccessKey ID ve AccessKey secret değerlerini kullanın. Bunların nasıl alınacağı hakkında daha fazla bilgi için bkz. AccessKey çifti alma.

    Kimlik bilgilerinin düz metin olarak açığa çıkmasını önlemek için bu örnekte AccessKey değerlerini belirtmek üzere ad alanı değişkenleri kullanılır. Daha fazla bilgi için bkz. Ad alanı değişkenleri.

    Not

    SLS ve Realtime Compute for Apache Flink hizmetleriniz farklı hesaplardaysa Flink hesabına SLS'ye yazma izni vermeniz gerekir. Daha fazla bilgi için bkz. Özel RAM ilkesi oluşturma. Aşağıdaki ilkelerden birini kullanın:

    • Kısıtlamasız SLS kapsamı

      {
      
          "Version": "1",
          "Statement": [
              {
                  "Effect": "Allow",
                  "Action": [
                      "log:Get*",
                      "log:PostLogStoreLogs"
                  ],
                  "Resource": "*"
              }
          ]
      }
    • Belirli bir SLS kaynağıyla sınırlı. Örnek:

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

    YOUR-SLS-ACCESSKEYSECRET

    flushIntervalSeconds

    Günlük verilerinin depolama hedefine yazılma aralığı (saniye cinsinden).

    flushIntervalEventCount

    Veriler hedefe yazılmadan önce arabelleğe alınacak günlük girişi sayısı.

    Not

    Hem flushIntervalSeconds hem de flushIntervalEventCount yapılandırılırsa yazma işlemi, hangi eşik karşılanırsa o anda tetiklenir.

    Not

    JAR işinde iş kodu, Flink günlük sistemi sınıf yükleyicisinden farklı olan ayrı bir sınıf yükleyici tarafından yüklenir. Varsayılan olarak günlük eklentisi sınıf yükleyicilerini katı bir şekilde doğrular (isInitStrictly varsayılan olarak true değerine sahiptir); bu nedenle bu tür günlükler atlanır ve SLS'e aktarılmaz. JAR işindeki iş kodu tarafından yazdırılan günlükleri toplamak için SLS Appender'a aşağıdaki yapılandırmayı ekleyin:

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

    Kafka

    Not

    Kafka kümeleri için Kerberos kimlik doğrulaması şu anda desteklenmemektedir.

    • Önkoşullar

      Realtime Compute for Apache Flink bir KafkaAppender günlük eklentisi sağlar. Flink bu eklentiyi bir eklenti sınıf yükleyicisi kullanarak yüklediğinden uygulamanızın eklentiyi başarıyla yükleyebilmesi için paket yolunu açıkça belirtmeniz gerekir. Bunu aşağıdaki yollardan biriyle yapabilirsiniz:

      • İş şablonu yapılandırma (bu çalışma alanındaki tüm dağıtımları etkiler)

        Konsolun Configurations sayfasında aşağıdaki kodu other configuration alanına ekleyin.

        plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
      • Tek bir dağıtım yapılandırma (yalnızca geçerli dağıtımı etkiler)

        Deployments sayfasında hedef dağıtımın adını tıklayın. Configuration sekmesinde, Parameters bölümündeki other configuration alanına aşağıdaki kodu ekleyin.

        plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
    • Günlük yapılandırması

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


      Parametre


      Açıklama


      YOUR-TOPIC-NAME


      Hedef Kafka konu adı.


      YOUR-KAFKA-BOOTSTRAP-SERVERS


      Hedef Kafka broker adresleri.


      YOUR-ACKS-VALUE


      Üreticinin yazma işlemini başarılı olarak kabul etmesi için mesajı onaylaması gereken bölüm çoğaltma sayısı. Daha fazla bilgi için bkz. acks.


      YOUR-BUFFER-MEMORY-SIZE


      Üretici arabellek boyutu (bayt cinsinden).


      YOUR-RETRIES-NUMBER


      Gönderim hatasından sonraki yeniden deneme sayısı.


      YOUR-COMPRESSION-TYPE


      Üreticinin mesajlar için kullandığı sıkıştırma türü. Geçerli değerler arasında none, gzip, snappy, lz4 ve zstd bulunur.

  3. Save seçeneğini tıklayın.

  4. Sayfanın üst kısmında Start seçeneğini tıklayın.

Ad alanı için günlük dışa aktarımını yapılandırın

Bir ad alanındaki tüm işler için varsayılan günlük dışa aktarma hedefini OSS, SLS veya Kafka olarak yapılandırmak üzere bir şablon kullanabilirsiniz.

Önemli
  • Bu şablonu yapılandırdığınızda bu ad alanındaki tüm yeni işlerin günlükleri OSS, SLS veya Kafka'ya aktarılır.

  • Flink eklenti sınıf yükleyicisi Realtime Compute for Apache Flink tarafından sağlanan KafkaAppender günlük eklentisini yükler. Bu eklentiyi kullanabilmeniz için Flink uygulamanızın eklentiyi yükleyebilmesine olanak tanımak amacıyla sınıf yolunu belirtmeniz gerekir. Bunu yapmak için Realtime Compute for Apache Flink geliştirme konsolunda Configurations sayfasına gidin ve aşağıdaki kodu Other Configuration bölümüne ekleyin.

    plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
  1. İş günlük şablonu yapılandırma sayfasına gidin.

    1. Realtime Compute for Apache Flink konsolunda oturum açın.

    2. Hedef çalışma alanının Actions sütununda Console seçeneğini tıklayın. Realtime Compute for Apache Flink geliştirme konsolunda sayfanın üst kısmından hedef namespace değerini seçin.

    3. Sol gezinme bölmesinde O&M > Configurations seçeneklerine gidin.

    4. Deployment Defaults sekmesinde bir iş türü seçin.

    5. Logging bölümünde Logging Profile değerini Custom Template olarak ayarlayın.

  2. Ad alanındaki tüm işler için günlük dışa aktarma hedefini yapılandırın.

    Gerekli kod için bkz. Tek bir iş için günlük dışa aktarımını yapılandırma (XML).

  3. save changes seçeneğini tıklayın.

Düzey tabanlı günlük çıkışlarını yapılandırın

Her ekleyici için günlük düzeyine göre farklı filtreleme kuralları yapılandırmak üzere Log4j2 ThresholdFilter öğesini kullanabilirsiniz. Bu yapılandırma birkaç avantaj sunar:

  • Esneklik: Farklı harici depolama hedefleri için farklı günlük düzeyleri ayarlayabilirsiniz.

  • Verimlilik: Gereksiz günlük işleme ve aktarımını azaltarak sistem performansını artırır.

  • Netlik: Ayrı yapılandırmalar günlük akışını netleştirir ve düzey yönetimini basitleştirir.

Şu adımları izleyin:

  1. logging bölümünde Logging Profile değerini Custom Template olarak ayarlayın.

  2. Günlük çıkışını yapılandırın.

    Aşağıdaki örnekte sistemin INFO düzeyi ve üzeri günlükleri Realtime Compute for Apache Flink geliştirme konsoluna gönderirken yalnızca ERROR düzeyi ve üzeri günlükleri Simple Log Service (SLS) hizmetine göndermesi için nasıl yapılandırılacağı gösterilmektedir.

    <?xml version="1.0" encoding="UTF-8"?>
    <Configuration xmlns="http://logging.apache.org/log4j/2.0/config" strict="true" status="WARN">
      <Appenders>
        <!-- Console Appender configured to output logs of INFO level and higher -->
        <Appender name="StdOut" type="Console">
          <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
          <Layout type="PatternLayout" pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" charset="UTF-8"/>
        </Appender>
        <!-- RollingFile Appender logs messages at the root logger's level (INFO) and higher -->
        <Appender name="RollingFile" type="RollingFile">
          <!-- Configuration remains unchanged -->
          <!-- ... -->
        </Appender>
        <!-- SLS Appender configured to output only ERROR level and above logs -->
        <Appender name="SLS" type="SLS">
          <ThresholdFilter level="ERROR" onMatch="ACCEPT" onMismatch="DENY"/>
          <Layout type="PatternLayout" pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" charset="UTF-8"/>
          <!-- SLS specific properties -->
          <Property name="namespace">YOUR_NAMESPACE</Property>
          <Property name="project">YOUR_SLS_PROJECT</Property>
          <Property name="logStore">YOUR_SLS_LOGSTORE</Property>
          <Property name="endpoint">YOUR_SLS_ENDPOINT</Property>
          <!-- Access credentials and other properties -->
          <!-- ... -->
        </Appender>
        <!-- Other Appenders definitions remain unchanged -->
        <!-- ... -->
      </Appenders>
      <Loggers>
        <!-- Define specific loggers that use the StdOut and SLS appenders -->
        <Logger name="StdOutLogger" level="INFO" additivity="false">
          <AppenderRef ref="StdOut"/>
        </Logger>
        <Logger name="SLSLogger" level="ERROR" additivity="false">
          <AppenderRef ref="SLS"/>
        </Logger>
        <!-- Other Loggers definitions with their specific configurations -->
        <!-- ... -->
        <!-- The Root logger sends output to StdOut and RollingFile, but not SLS, to avoid duplicate logs. -->
        <Root level="INFO">
          <AppenderRef ref="StdOut"/>
          <AppenderRef ref="RollingFile"/>
          <!-- Exclude SLS from Root to prevent duplicate logging in case of other loggers -->
        </Root>
      </Loggers>
    </Configuration>

    Bu yapılandırmada:

    • Console Appender: INFO düzeyi veya üzeri günlüklerin geliştirme konsoluna aktarılmasını sağlamak için ThresholdFilter kullanır.

    • SLS Appender, yalnızca ERROR düzeyi ve üzeri günlükleri SLS'e göndermek için bir ThresholdFilter kullanır. SLS Appender özellikleriyle ilgili ayrıntılar için bkz. SLS'e Çıkışı Yapılandırma. YOUR_NAMESPACE ve YOUR_SLS_PROJECT gibi yer tutucuları gerçek SLS proje bilgilerinizle değiştirin.

      Not

      SLS Appender türü SLS değil de özel bir ekleyici ise doğru türü kullandığınızdan ve ekleyici sınıfının SLS'ye bağlanmak için gereken mantığa sahip olduğundan emin olun.

    • StdOutLogger ve SLSLogger: Günlükleri sırasıyla yalnızca StdOut Appender ve SLS Appender'a gönderirler ve her birinin kendi günlük düzeyi kısıtlamaları vardır.

    • Root Logger: Bu günlük kaydedici bir StdOut Appender ve bir RollingFile Appender ile yapılandırılmıştır ancak bir SLS Appender içermez. Belirli bir günlük kaydedici zaten bu şekilde yapılandırılmışsa SLS'e yinelenen günlüklerin gönderilmesini önlemek için bu yapılır.

    Log4j yapılandırma parametreleri hakkında daha fazla bilgi için bkz. Apache Log4j.

İlgili konular