Menggunakan log pipeline

Anda dapat menggunakan infrastruktur logging bawaan Apache Beam SDK untuk mencatat informasi saat menjalankan pipeline. Anda dapat menggunakan konsol Google Cloud untuk memantau informasi logging selama dan setelah pipeline berjalan.

Menambahkan pesan log ke pipeline

Java

Apache Beam SDK untuk Java merekomendasikan agar Anda mencatat pesan pekerja melalui library Simple Logging Facade for Java (SLF4J) open source. Apache Beam SDK untuk Java menerapkan infrastruktur logging yang diperlukan sehingga kode Java Anda hanya perlu mengimpor SLF4J API. Kemudian, kode ini membuat instance Logger untuk mengaktifkan logging pesan dalam kode pipeline Anda.

Untuk kode dan/atau library yang sudah ada, Apache Beam SDK untuk Java menyiapkan infrastruktur logging tambahan. Pesan log yang dihasilkan oleh library logging berikut untuk Java akan direkam:

Python

Apache Beam SDK untuk Python menyediakan paket library logging, yang memungkinkan pekerja pipeline menghasilkan pesan log. Untuk menggunakan fungsi library, Anda harus mengimpor library:

import logging

Go

Apache Beam SDK untuk Go menyediakan paket library log, yang memungkinkan pekerja pipeline menghasilkan pesan log. Untuk menggunakan fungsi library, Anda harus mengimpor library:

import "github.com/apache/beam/sdks/v2/go/pkg/beam/log"

Contoh kode pesan log pekerja

Java

Contoh berikut menggunakan SLF4J untuk logging Dataflow. Untuk mempelajari lebih lanjut cara mengonfigurasi SLF4J untuk logging Dataflow, lihat artikel Tips Java.

Contoh WordCount Apache Beam dapat diubah untuk menghasilkan pesan log saat kata "love" ditemukan di baris teks yang diproses. Kode yang ditambahkan ditunjukkan dalam cetak tebal pada contoh berikut (kode di sekitarnya disertakan untuk konteks).

 package org.apache.beam.examples;
 // Import SLF4J packages.
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 ...
 public class WordCount {
   ...
   static class ExtractWordsFn extends DoFn<String, String> {
     // Instantiate Logger.
     // Suggestion: As shown, specify the class name of the containing class
     // (WordCount).
     private static final Logger LOG = LoggerFactory.getLogger(WordCount.class);
     ...
     @ProcessElement
     public void processElement(ProcessContext c) {
       ...
       // Output each word encountered into the output PCollection.
       for (String word : words) {
         if (!word.isEmpty()) {
           c.output(word);
         }
         // Log INFO messages when the word "love" is found.
         if(word.toLowerCase().equals("love")) {
           LOG.info("Found " + word.toLowerCase());
         }
       }
     }
   }
 ... // Remaining WordCount example code ...

Python

Contoh wordcount.py Apache Beam dapat diubah untuk menghasilkan pesan log saat kata "love" ditemukan dalam baris teks yang diproses.

# import Python logging module.
import logging

class ExtractWordsFn(beam.DoFn):
  def process(self, element):
    words = re.findall(r'[A-Za-z\']+', element)
    for word in words:
      yield word

      if word.lower() == 'love':
        # Log using the root logger at info or higher levels
        logging.info('Found : %s', word.lower())

# Remaining WordCount example code ...

Go

Contoh wordcount.go Apache Beam dapat diubah untuk menghasilkan pesan log saat kata "love" ditemukan di baris teks yang diproses.

func (f *extractFn) ProcessElement(ctx context.Context, line string, emit func(string)) {
    for _, word := range wordRE.FindAllString(line, -1) {
        // increment the counter for small words if length of words is
        // less than small_word_length
        if strings.ToLower(word) == "love" {
            log.Infof(ctx, "Found : %s", strings.ToLower(word))
        }

        emit(word)
    }
}

// Remaining Wordcount example

Java

Jika pipeline WordCount yang dimodifikasi dijalankan secara lokal menggunakan DirectRunner default dengan output yang dikirim ke file lokal (--output=./local-wordcounts), output konsol akan menyertakan pesan log yang ditambahkan:

INFO: Executing pipeline using the DirectRunner.
...
Feb 11, 2015 1:13:22 PM org.apache.beam.examples.WordCount$ExtractWordsFn processElement
INFO: Found love
Feb 11, 2015 1:13:22 PM org.apache.beam.examples.WordCount$ExtractWordsFn processElement
INFO: Found love
Feb 11, 2015 1:13:22 PM org.apache.beam.examples.WordCount$ExtractWordsFn processElement
INFO: Found love
...
INFO: Pipeline execution complete.

Secara default, hanya baris log yang ditandai INFO dan yang lebih tinggi yang akan dikirim ke Cloud Logging. Jika Anda ingin mengubah perilaku ini, lihat Menetapkan Level Log Pekerja Pipeline.

Python

Jika pipeline WordCount yang dimodifikasi dijalankan secara lokal menggunakan DirectRunner default dengan output yang dikirim ke file lokal (--output=./local-wordcounts), output konsol akan menyertakan pesan log yang ditambahkan:

INFO:root:Found : love
INFO:root:Found : love
INFO:root:Found : love

Secara default, hanya baris log yang ditandai INFO dan yang lebih tinggi yang akan dikirim ke Cloud Logging.

Go

Jika pipeline WordCount yang dimodifikasi dijalankan secara lokal menggunakan DirectRunner default dengan output yang dikirim ke file lokal (--output=./local-wordcounts), output konsol akan menyertakan pesan log yang ditambahkan:

2022/05/26 11:36:44 Found : love
2022/05/26 11:36:44 Found : love
2022/05/26 11:36:44 Found : love

Secara default, hanya baris log yang ditandai INFO dan yang lebih tinggi yang akan dikirim ke Cloud Logging.

Mengontrol volume log

Anda juga dapat mengurangi volume log yang dihasilkan dengan mengubah level log pipeline. Jika Anda tidak ingin terus menyerap sebagian atau semua log Dataflow, tambahkan pengecualian Logging untuk mengecualikan log Dataflow. Kemudian, ekspor log ke tujuan lain seperti BigQuery, Cloud Storage, atau Pub/Sub. Untuk mengetahui informasi selengkapnya, lihat Mengontrol penyerapan log Dataflow.

Batas dan throttling logging

Pesan log pekerja dibatasi hingga 15.000 pesan setiap 30 detik, per pekerja. Jika batas ini tercapai, satu pesan log pekerja akan ditambahkan yang menyatakan bahwa logging dibatasi: 

Throttling logger worker. It used up its 30s quota for logs in only 12.345s
Tidak ada lagi pesan yang dicatat ke dalam log hingga interval 30 detik berakhir. Batas ini digunakan bersama oleh pesan log yang dihasilkan oleh Apache Beam SDK dan kode pengguna.

Penyimpanan dan retensi log

Log operasional disimpan di bucket log _Default. Nama layanan logging API adalah dataflow.googleapis.com. Untuk mengetahui informasi selengkapnya tentang jenis resource dan layanan yang dimonitor Google Cloud yang digunakan di Cloud Logging, lihat Resource dan layanan yang dimonitor.

Untuk mengetahui detail tentang berapa lama entri log disimpan oleh Logging, lihat informasi retensi di Kuota dan batas: Periode retensi log.

Untuk informasi tentang cara melihat log operasional, lihat Memantau dan melihat log pipeline.

Memantau dan melihat log pipeline

Saat menjalankan pipeline di layanan Dataflow, Anda dapat menggunakan antarmuka pemantauan Dataflow untuk melihat log yang dikeluarkan oleh pipeline.

Contoh log pekerja Dataflow

Pipeline WordCount yang dimodifikasi dapat dijalankan di cloud dengan opsi berikut:

Java

--project=WordCountExample
--output=gs://<bucket-name>/counts
--runner=DataflowRunner
--tempLocation=gs://<bucket-name>/temp
--stagingLocation=gs://<bucket-name>/binaries

Python

--project=WordCountExample
--output=gs://<bucket-name>/counts
--runner=DataflowRunner
--staging_location=gs://<bucket-name>/binaries

Go

--project=WordCountExample
--output=gs://<bucket-name>/counts
--runner=DataflowRunner
--staging_location=gs://<bucket-name>/binaries

Lihat log

Karena pipeline cloud WordCount menggunakan eksekusi pemblokiran, pesan konsol merupakan output selama eksekusi pipeline. Setelah tugas dimulai, link ke halaman konsol Google Cloud akan ditampilkan ke konsol, diikuti dengan ID tugas pipeline:

INFO: To access the Dataflow monitoring console, please navigate to
https://console.developers.google.com/dataflow/job/2017-04-13_13_58_10-6217777367720337669
Submitted job: 2017-04-13_13_58_10-6217777367720337669

URL konsol mengarah ke antarmuka pemantauan Dataflow dengan halaman ringkasan untuk tugas yang dikirim. Tampilan ini menampilkan grafik eksekusi dinamis di kiri, dengan informasi ringkasan di sebelah kanan. Klik di panel bawah untuk meluaskan panel log.

Panel log secara default menampilkan Log Tugas yang melaporkan status tugas secara keseluruhan. Anda dapat memfilter pesan yang muncul di panel log dengan mengklik Info dan Filter log.

Memilih langkah pipeline dalam grafik akan mengubah tampilan menjadi Log Langkah yang dihasilkan oleh kode Anda dan kode yang dihasilkan yang berjalan di langkah pipeline.

Untuk kembali ke Log Tugas, hapus langkah dengan mengklik di luar grafik atau menggunakan tombol Batalkan pilihan langkah di panel samping kanan.

Membuka Logs Explorer

Untuk membuka Logs Explorer dan memilih jenis log yang berbeda, di panel log, klik Lihat di Logs Explorer (tombol link eksternal).

Di Logs Explorer, untuk melihat panel dengan jenis log yang berbeda, klik tombol Log fields.

Di halaman Logs Explorer, kueri dapat memfilter log menurut langkah tugas atau menurut jenis log. Untuk menghapus filter, klik tombol Tampilkan kueri, lalu edit kueri.

Untuk melihat semua log yang tersedia untuk tugas, ikuti langkah-langkah berikut:

  1. Di kolom Query, masukkan kueri berikut:

    resource.type="dataflow_step"
resource.labels.job_id="JOB_ID"

    Ganti JOB_ID dengan ID tugas Anda.

  2. Klik Run query.

  3. Jika Anda menggunakan kueri ini dan tidak melihat log untuk tugas, klik Edit time.

  4. Sesuaikan waktu mulai dan waktu akhir, lalu klik Terapkan.

Jenis log

Logs Explorer juga menyertakan log infrastruktur untuk pipeline Anda. Gunakan log error dan peringatan untuk mendiagnosis masalah pipeline yang diamati. Error dan peringatan dalam log infrastruktur yang tidak berkorelasi dengan masalah pipeline tidak selalu menunjukkan adanya masalah.

Berikut adalah ringkasan berbagai jenis log yang tersedia untuk dilihat dari halaman Logs Explorer:

  • Log job-message berisi pesan tingkat tugas yang dihasilkan oleh berbagai komponen Dataflow. Contohnya mencakup konfigurasi penskalaan otomatis, saat pekerja memulai atau menonaktifkan, progres pada langkah tugas, dan error tugas. Error tingkat pekerja yang berasal dari error kode pengguna dan yang ada di log pekerja juga di-propagate ke log pesan tugas.
  • Log worker dihasilkan oleh pekerja Dataflow. Pekerja melakukan sebagian besar pekerjaan pipeline (misalnya, menerapkan ParDo ke data). Log Pekerja berisi pesan yang dicatat ke dalam log oleh kode dan Dataflow Anda.
  • Log worker-startup ada di sebagian besar tugas Dataflow dan dapat merekam pesan yang terkait dengan proses startup. Proses startup mencakup mendownload jar tugas dari Cloud Storage, lalu memulai pekerja. Jika ada masalah saat memulai pekerja, log ini adalah tempat yang tepat untuk melihatnya.
  • Log shuffler berisi pesan dari pekerja yang menggabungkan hasil operasi pipeline paralel.
  • Log system berisi pesan dari sistem operasi host VM pekerja. Dalam beberapa skenario, peristiwa ini mungkin merekam error proses atau peristiwa kehabisan memori (OOM).
  • Log docker dan kubelet berisi pesan yang terkait dengan teknologi publik ini, yang digunakan di pekerja Dataflow.
  • Log nvidia-mps berisi pesan tentang operasi NVIDIA Multi-Process Service (MPS).

Menetapkan level log pekerja pipeline

Java

Level logging SLF4J default yang ditetapkan pada pekerja oleh Apache Beam SDK untuk Java adalah INFO. Semua pesan log INFO atau yang lebih tinggi (INFO, WARN, ERROR) akan ditampilkan. Anda dapat menetapkan level log default yang berbeda untuk mendukung level logging SLF4J yang lebih rendah (TRACE atau DEBUG) atau menetapkan level log yang berbeda untuk paket class yang berbeda dalam kode Anda.

Opsi pipeline berikut disediakan agar Anda dapat menetapkan level log pekerja dari command line atau secara terprogram:

  • --defaultSdkHarnessLogLevel=<level>: gunakan opsi ini untuk menetapkan semua logger pada level default yang ditentukan. Misalnya, opsi command line berikut akan mengganti tingkat log INFO Dataflow default, dan menetapkannya ke DEBUG:
    --defaultSdkHarnessLogLevel=DEBUG
  • --sdkHarnessLogLevelOverrides={"<package or class>":"<level>"}: gunakan opsi ini untuk menetapkan tingkat logging untuk paket atau class yang ditentukan. Misalnya, untuk mengganti level log pipeline default untuk paket org.apache.beam.runners.dataflow, dan menetapkannya ke TRACE:
    --sdkHarnessLogLevelOverrides='{"org.apache.beam.runners.dataflow":"TRACE"}'
    Untuk membuat beberapa penggantian, berikan peta JSON:
    (--sdkHarnessLogLevelOverrides={"<package/class>":"<level>","<package/class>":"<level>",...}).
  • Opsi pipeline defaultSdkHarnessLogLevel dan sdkHarnessLogLevelOverrides tidak didukung dengan pipeline yang menggunakan Apache Beam SDK versi 2.50.0 dan yang lebih lama tanpa Runner v2. Dalam hal ini, gunakan opsi pipeline --defaultWorkerLogLevel=<level> dan --workerLogLevelOverrides={"<package or class>":"<level>"}. Untuk membuat beberapa penggantian, berikan peta JSON:
    (--workerLogLevelOverrides={"<package/class>":"<level>","<package/class>":"<level>",...})

Contoh berikut menetapkan opsi logging pipeline secara terprogram dengan nilai default yang dapat diganti dari command line:

 PipelineOptions options = ...
 SdkHarnessOptions loggingOptions = options.as(SdkHarnessOptions.class);
 // Overrides the default log level on the worker to emit logs at TRACE or higher.
 loggingOptions.setDefaultSdkHarnessLogLevel(LogLevel.TRACE);
 // Overrides the Foo class and "org.apache.beam.runners.dataflow" package to emit logs at WARN or higher.
 loggingOptions.getSdkHarnessLogLevelOverrides()
     .addOverrideForClass(Foo.class, LogLevel.WARN)
     .addOverrideForPackage(Package.getPackage("org.apache.beam.runners.dataflow"), LogLevel.WARN);

Python

Tingkat logging default yang ditetapkan pada pekerja oleh Apache Beam SDK untuk Python adalah INFO. Semua pesan log INFO atau yang lebih tinggi (INFO, WARNING, ERROR, CRITICAL) akan ditampilkan. Anda dapat menetapkan level log default yang berbeda untuk mendukung level logging yang lebih rendah (DEBUG) atau menetapkan level log yang berbeda untuk modul yang berbeda dalam kode Anda.

Dua opsi pipeline disediakan untuk memungkinkan Anda menetapkan level log pekerja dari command line atau secara terprogram:

  • --default_sdk_harness_log_level=<level>: gunakan opsi ini untuk menetapkan semua logger pada level default yang ditentukan. Misalnya, opsi command line berikut mengganti tingkat log INFO Dataflow default, dan menetapkannya ke DEBUG:
    --default_sdk_harness_log_level=DEBUG
  • --sdk_harness_log_level_overrides={\"<module>\":\"<level>\"}: gunakan opsi ini untuk menetapkan level logging untuk modul yang ditentukan. Misalnya, untuk mengganti tingkat log pipeline default untuk modul apache_beam.runners.dataflow, dan menetapkannya ke DEBUG:
    --sdk_harness_log_level_overrides={\"apache_beam.runners.dataflow\":\"DEBUG\"}
    Untuk membuat beberapa penggantian, berikan peta JSON:
    (--sdk_harness_log_level_overrides={\"<module>\":\"<level>\",\"<module>\":\"<level>\",...}).

Contoh berikut menggunakan class WorkerOptions untuk menetapkan opsi logging pipeline secara terprogram yang dapat diganti dari command line:

  from apache_beam.options.pipeline_options import PipelineOptions, WorkerOptions

  pipeline_args = [
    '--project=PROJECT_NAME',
    '--job_name=JOB_NAME',
    '--staging_location=gs://STORAGE_BUCKET/staging/',
    '--temp_location=gs://STORAGE_BUCKET/tmp/',
    '--region=DATAFLOW_REGION',
    '--runner=DataflowRunner'
  ]

  pipeline_options = PipelineOptions(pipeline_args)
  worker_options = pipeline_options.view_as(WorkerOptions)
  worker_options.default_sdk_harness_log_level = 'WARNING'

  # Note: In Apache Beam SDK 2.42.0 and earlier versions, use ['{"apache_beam.runners.dataflow":"WARNING"}']
  worker_options.sdk_harness_log_level_overrides = {"apache_beam.runners.dataflow":"WARNING"}

  # Pass in pipeline options during pipeline creation.
  with beam.Pipeline(options=pipeline_options) as pipeline:

Ganti kode berikut:

  • PROJECT_NAME: nama project
  • JOB_NAME: nama tugas
  • STORAGE_BUCKET: nama Cloud Storage
  • DATAFLOW_REGION: region tempat Anda ingin men-deploy tugas Dataflow

    Flag --region akan menggantikan region default yang ditetapkan di server metadata, klien lokal, atau variabel lingkungan.

Go

Fitur ini tidak tersedia di Apache Beam SDK untuk Go.

Melihat log tugas BigQuery yang diluncurkan

Saat menggunakan BigQuery di pipeline Dataflow, tugas BigQuery akan diluncurkan untuk melakukan berbagai tindakan atas nama Anda. Tindakan ini dapat mencakup memuat data, mengekspor data, dan sebagainya. Untuk tujuan pemecahan masalah dan pemantauan, antarmuka pemantauan Dataflow memiliki informasi tambahan tentang tugas BigQuery ini yang tersedia di panel Logs.

Informasi tugas BigQuery yang ditampilkan di panel Logs disimpan dan dimuat dari tabel sistem BigQuery. Biaya penagihan akan dikenakan saat tabel BigQuery yang mendasarinya dikueri.

Melihat detail tugas BigQuery

Untuk melihat informasi tugas BigQuery, pipeline Anda harus menggunakan Apache Beam 2.24.0 atau yang lebih baru.

Untuk mencantumkan tugas BigQuery, buka tab BigQuery Jobs dan pilih lokasi tugas BigQuery. Selanjutnya, klik Muat Tugas BigQuery dan konfirmasi dialog. Setelah kueri selesai, daftar tugas akan ditampilkan.

Tombol Muat Tugas BigQuery di tabel informasi tugas BigQuery

Informasi dasar tentang setiap tugas diberikan, termasuk ID tugas, jenis, durasi, dan sebagainya.

Tabel yang menampilkan tugas BigQuery yang dijalankan selama eksekusi tugas pipeline saat ini.

Untuk informasi yang lebih mendetail tentang tugas tertentu, klik Command line di kolom More Info.

Di jendela modal untuk command line, salin perintah bq jobs describe dan jalankan secara lokal atau di Cloud Shell.

gcloud alpha bq jobs describe BIGQUERY_JOB_ID

Perintah bq jobs describe menghasilkan JobStatistics, yang memberikan detail lebih lanjut yang berguna saat mendiagnosis tugas BigQuery yang lambat atau macet.

Atau, saat Anda menggunakan BigQueryIO dengan kueri SQL, tugas kueri akan dikeluarkan. Untuk melihat kueri SQL yang digunakan oleh tugas, klik Lihat kueri di kolom Info Selengkapnya.