Google menggunakan teknologi AI untuk menerjemahkan konten ke dalam bahasa pilihan Anda. Terjemahan AI mungkin mengandung kesalahan.

Menggunakan pelacakan terdistribusi OpenTelemetry di Dataflow

Dalam pipeline streaming terdistribusi, Anda mungkin kesulitan mengidentifikasi tahap tertentu, panggilan layanan eksternal, atau pengacakan pekerja yang menyebabkan lonjakan latensi hanya dengan menggunakan metrik gabungan. Dengan menggunakan pelacakan terdistribusi OpenTelemetry di Dataflow, Anda dapat melacak setiap elemen secara end-to-end di seluruh transformasi pipeline, batas jaringan, dan layanan terintegrasi dengan satu ID Trace. Visibilitas ini membantu Anda menentukan bottleneck performa, mengoptimalkan biaya eksekusi pipeline, dan memecahkan masalah workload streaming yang kompleks.

Gunakan informasi berikut untuk mengaktifkan, mengonfigurasi, dan menggunakan distributed tracing OpenTelemetry di pipeline Apache Beam Anda.

Kasus penggunaan

Pelacakan terdistribusi di Dataflow sangat berguna dalam skenario berikut:

Beban kerja streaming: Lacak waktu pemrosesan per elemen dan identifikasi tahap yang menyebabkan lonjakan jeda atau latensi.
Pelacakan panggilan layanan eksternal dalam kode pengguna: Mengukur durasi eksekusi saat pipeline Anda memanggil database eksternal, microservice, atau API AI generatif di dalam kode pengguna (seperti dalam DoFn). Misalnya, Anda dapat mengukur waktu untuk panggilan ke database eksternal (seperti Spanner atau Bigtable), microservice (menggunakan HTTP atau gRPC), atau API AI generatif (seperti Gemini). Untuk mengetahui informasi selengkapnya, lihat Menulis rekaman aktivitas kustom di DoFn dan Mengintegrasikan dengan library klien.
Pengoptimalan biaya dan performa: Menampilkan durasi operasi tertentu dalam instance PTransform di bawah beban dunia nyata untuk mengidentifikasi serialisasi yang tidak efisien, kueri lambat, atau pertentangan thread pekerja.
Pipeline alur kerja dan agentic yang kompleks: Dapatkan visibilitas ke dalam pipeline dengan beberapa cabang, loop agen iteratif, dan penggabungan untuk menentukan transformasi atau cabang mana yang paling lama.

Batasan

Pelacakan terdistribusi di Dataflow memiliki batasan berikut:

Fitur ini dirancang dan diverifikasi hanya untuk pipeline streaming.
Untuk menggunakan fitur ini, Anda harus menggunakan Beam SDK untuk Java (versi 2.76.0 atau yang lebih baru).
Propagasi konteks lintas tahap memerlukan Streaming Java Runner (sebelumnya disebut Runner v1). Portable Runner (sebelumnya disebut Runner v2) mereset konteks rekaman aktivitas di batas pengacakan.
Pelacakan OpenTelemetry bawaan di konektor I/O Beam didukung hanya untuk konektor berikut: Pub/Sub, Apache Kafka, dan aliran perubahan Spanner. Untuk Apache Kafka, pelacakan hanya didukung saat Anda menggunakan konektor KafkaIO secara langsung. Fitur ini tidak didukung saat Anda menggunakan I/O Terkelola untuk Kafka. Konektor I/O lainnya (seperti BigtableIO) tidak memiliki pelacakan konektor bawaan; untuk melacak operasi dengan layanan ini, lakukan instrumentasi panggilan Anda di dalam kode pengguna (seperti di DoFn).
Operasi yang menggabungkan beberapa elemen menjadi satu tidak menyebarkan satu konteks rekaman aktivitas. Hal ini mencakup operasi seperti GroupByKey, CoGroupByKey, dan Combine.
Konteks tidak diteruskan antara penyetelan timer (Timer.set()) dan eksekusi metode callback @OnTimer.

Prasyarat

Untuk menggunakan pelacakan terdistribusi di Dataflow, pastikan lingkungan Anda memenuhi persyaratan berikut:

Tugas Anda harus memenuhi persyaratan berikut:
- Adalah pipeline streaming
- Menggunakan Streaming Java Runner (sebelumnya disebut Runner v1)
- Menggunakan Apache Beam Java SDK versi 2.76.0 atau yang lebih baru
Jika Anda mengekspor rekaman aktivitas ke Cloud Trace, aktifkan Telemetry API ( telemetry.googleapis.com) di project Google Cloud Anda.

Peran yang diperlukan untuk mengaktifkan API
Untuk mengaktifkan API, Anda memerlukan izin serviceusage.services.enable. Jika Anda membuat project, kemungkinan Anda sudah memiliki izin ini melalui peran Pemilik (roles/owner). Jika tidak, Anda bisa mendapatkan izin ini melalui peran Admin Penggunaan Layanan (roles/serviceusage.serviceUsageAdmin). Pelajari cara memberikan peran.
Mengaktifkan API
Akun layanan pekerja harus memiliki peran Cloud Trace Agent (roles/cloudtrace.agent), yang mencakup izin cloudtrace.traces.patch.

Konsep inti

Pelacakan terdistribusi mengandalkan konsep OpenTelemetry berikut:

Trace: Representasi transaksi atau alur kerja lengkap saat mengalir melalui sistem terdistribusi. Trace terdiri dari satu atau beberapa span.
Span: Elemen penyusun dasar rekaman aktivitas, yang merepresentasikan satu unit kerja. Misalnya, eksekusi transformasi, panggilan metode, atau permintaan RPC.
Konteks: Status yang disebarkan di seluruh thread dan batas API. Hal ini mencakup ID Pelacakan dan ID Rentang.
Propagator: Mekanisme yang digunakan untuk menyuntikkan dan mengekstrak representasi konteks di seluruh batas jaringan. Misalnya, Konteks Trace W3C.
Pengekspor: Komponen yang mengirimkan data span yang dikumpulkan ke backend pelacakan, seperti Cloud Trace atau OpenTelemetry Collector yang dihosting sendiri.
Sampler: Mekanisme yang mengontrol volume rekaman aktivitas dengan menentukan rekaman aktivitas mana yang direkam dan diekspor. Misalnya, frekuensi sampling 1%.

Mengaktifkan pelacakan OpenTelemetry

Untuk mengaktifkan pelacakan OpenTelemetry dengan setelan default, konfigurasi opsi pipeline dan eksperimen yang diperlukan saat Anda mengirimkan tugas. Setelan default mengekspor ke Cloud Trace dengan sampling 1%.

Untuk mengubah nilai default, lihat Menetapkan dan mengubah properti pelacakan.

Konfigurasi opsi pipeline dan eksperimen saat Anda menjalankan tugas.
Tugas template non-flex

Teruskan tanda eksperimen berikut saat Anda mengirimkan tugas:
```
--experiments=enable_otel_defaults,element_metadata_supported,disable_portable_worker
```
Tugas template flex

Luluskan eksperimen menggunakan flag --additional-experiments dalam perintah gcloud:
```
gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://BUCKET_NAME/templates/TEMPLATE_NAME.json \
    --additional-experiments="enable_otel_defaults,element_metadata_supported,disable_portable_worker" \
    --region=REGION
```
Ganti kode berikut:
- JOB_NAME: Nama tugas Dataflow Anda.
- BUCKET_NAME: Bucket Cloud Storage yang berisi template Anda.
- TEMPLATE_NAME: Nama file Flex Template Anda.
- REGION: Google Cloud Region tempat Anda ingin menjalankan tugas.
Eksperimen ini mengonfigurasi perilaku berikut:
- enable_otel_defaults: Mengonfigurasi default OpenTelemetry standar, termasuk pengekspor Cloud Trace dan sampling 1%.
- element_metadata_supported: Mengaktifkan serialisasi dan propagasi metadata (seperti konteks rekaman aktivitas) di seluruh tahap pipeline.
- disable_portable_worker: Memaksa penggunaan Dataflow Streaming Java Runner (sebelumnya disebut Runner v1).

Jika Anda menggunakan Beam SDK untuk Java versi 2.76.0 atau 2.77.0, tambahkan dependensi OpenTelemetry yang diperlukan ke konfigurasi build Anda.

Karena masalah pada versi 2.76.0 dan 2.77.0, Anda harus menambahkan dependensi berikut ke file pom.xml agar rekaman aktivitas dapat diekspor. Untuk mengetahui informasi selengkapnya, lihat Mengelola dependensi pipeline.

<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
    <scope>compile</scope>
</dependency>

<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-otlp</artifactId>
    <version>1.62.0</version>
    <scope>compile</scope>
</dependency>

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-extensions-opentelemetry-gcp-auth-extension</artifactId>
    <scope>compile</scope>
</dependency>

Aktifkan pelacakan OpenTelemetry secara eksplisit pada konektor I/O yang didukung dalam kode pipeline Anda:
Pub/Sub

Untuk mengaktifkan pelacakan saat membaca dari atau menulis ke Pub/Sub:
```
// Reading from Pub/Sub
pipeline.apply("ReadFromPubSub",
    PubsubIO.readMessages()
        .withEnableOpenTelemetryTracing()
        .fromSubscription("projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID"));

// Writing to Pub/Sub
pipeline.apply("WriteToPubSub",
    PubsubIO.writeStrings()
        .withEnableOpenTelemetryTracing()
        .to("projects/PROJECT_ID/topics/TOPIC_ID"));
```
Ganti kode berikut:
- PROJECT_ID: ID Google Cloud project Anda.
- SUBSCRIPTION_ID: Nama langganan Pub/Sub Anda.
- TOPIC_ID: Nama topik Pub/Sub Anda.
Detail penerapan konteks:
- Publisher upstream: Saat memublikasikan pesan dari aplikasi eksternal, masukkan Konteks Trace W3C ke dalam atribut pesan Pub/Sub. Untuk mengetahui informasi selengkapnya, lihat Pelacakan OpenTelemetry Pub/Sub.
- Atribut: Beam mengharapkan dan mengisi atribut berikut:
  - googclient_traceparent: Membawa header traceparent W3C.
  - googclient_tracestate: Membawa header tracestate W3C.
- Untuk pemformatan header, lihat spesifikasi Konteks Trace W3C.
Apache Kafka

Catatan: Pelacakan untuk Kafka hanya didukung jika Anda menggunakan konektor KafkaIO secara langsung. Fitur ini tidak didukung saat Anda menggunakan I/O Terkelola untuk Kafka.

Untuk mengaktifkan pelacakan saat membaca dari atau menulis ke Apache Kafka:
```
// Reading from Kafka
pipeline.apply("ReadFromKafka",
    KafkaIO.<String, String>read()
        .withEnableOpenTelemetryTracing()
        // ... additional configuration
        );

// Writing to Kafka
pipeline.apply("WriteToKafka",
    KafkaIO.<String, String>write()
        .withEnableOpenTelemetryTracing()
        // ... additional configuration
        );
```
Detail penerapan konteks:
- KafkaIO menggunakan header rekaman Kafka standar untuk menyebarkan konteks.
- Library ini mengharapkan dan menyuntikkan header yang sesuai dengan spesifikasi W3C Trace Context:
  - traceparent: Mengidentifikasi konteks rekaman aktivitas masuk.
  - tracestate: Menyediakan metadata perutean dan pemfilteran tambahan khusus vendor.
Spanner

Untuk mengaktifkan pelacakan saat membaca dari aliran perubahan Spanner:
```
pipeline.apply("ReadChangeStream",
    SpannerIO.readChangeStream()
        .withSpannerConfig(spannerConfig)
        .withEnableOpenTelemetryTracing(true)
        .withChangeStreamName("CHANGE_STREAM_NAME"));
```
Ganti kode berikut:
- CHANGE_STREAM_NAME: Nama aliran data perubahan Spanner.
Detail pembuatan rentang:
- Mengaktifkan pelacakan akan memulai rekaman aktivitas dan rentang baru untuk setiap rekaman mutasi database yang dibaca oleh pembaca Pengambilan Data Perubahan (CDC).
- Siklus polling pembaca Spanner yang tidak menampilkan data baru juga dapat menghasilkan rentang aktivitas yang dapat difilter di UI Cloud Trace.

Menetapkan dan mengubah properti pelacakan

Saat Anda menentukan enable_otel_defaults, Dataflow akan menerapkan properti default berikut:

Properti	Nilai default	Deskripsi
`otel.traces.exporter`	`otlp`	Mengekspor rekaman aktivitas menggunakan OpenTelemetry Protocol (OTLP).
`otel.exporter.otlp.endpoint`	`https://telemetry.googleapis.com`	Menargetkan penerima OTLP Cloud Trace.
`google.cloud.project`	Project ID Anda	Project Google Cloud tempat rekaman aktivitas disimpan.
`otel.traces.sampler.arg`	`0.01` (1%)	Mencatat 1 dari setiap 100 rekaman aktivitas.
`otel.service.name`	`options.getAppName()`	Nama layanan yang terkait dengan rentang, yang berasal dari opsi pipeline `appName`.

Untuk menyesuaikan konfigurasi OpenTelemetry, teruskan daftar properti yang dipisahkan titik koma ke opsi pipeline --openTelemetryProperties.

Contoh berikut mengonfigurasi semua properti default secara eksplisit:

--openTelemetryProperties="otel.traces.exporter=otlp;otel.exporter.otlp.endpoint=https://telemetry.googleapis.com;google.cloud.project=PROJECT_ID;otel.traces.sampler.arg=0.01;otel.service.name=CUSTOM_SERVICE_NAME;otel.java.global-autoconfigure.enabled=true"

Menetapkan nama layanan kustom

Secara default, nama layanan rekaman aktivitas berasal dari opsi appName pipeline (--appName=YourPipelineName). Anda dapat mengganti nilai ini dengan menyetel properti otel.service.name:

--openTelemetryProperties="otel.service.name=CUSTOM_SERVICE_NAME"

Mengubah frekuensi sampling dan properti

Menetapkan frekuensi sampling yang lebih tinggi akan memberi Anda informasi pelacakan yang lebih andal. Namun, frekuensi sampling yang tinggi (seperti 1.0 atau always_on) akan meningkatkan volume data rekaman aktivitas yang dikirim ke Cloud Trace secara signifikan, yang dapat meningkatkan biaya Google Cloud . Gunakan frekuensi sampel tinggi hanya untuk proses debug jangka pendek dan kurangi frekuensi sampel untuk workload produksi.

Contoh berikut mengonfigurasi rasio pengambilan sampel secara eksplisit. Dalam hal ini, Anda harus memberikan daftar lengkap argumen:

--openTelemetryProperties="otel.traces.exporter=otlp;otel.exporter.otlp.endpoint=https://telemetry.googleapis.com;google.cloud.project=PROJECT_ID;otel.traces.sampler.arg=0.02;otel.service.name=CUSTOM_SERVICE_NAME;otel.java.global-autoconfigure.enabled=true"

Mengekspor ke backend pihak ketiga

Untuk mengirim trace ke backend tracing terkelola eksternal atau OpenTelemetry Collector yang dihosting sendiri, bukan Cloud Trace:

Kecualikan eksperimen berikut: --experiments=enable_otel_defaults.

Mengonfigurasi properti pengekspor secara manual menggunakan --openTelemetryProperties:

--openTelemetryProperties="otel.traces.exporter=otlp;otel.exporter.otlp.endpoint=http://COLLECTOR_HOST:4317;otel.traces.sampler=parentbased_always_on"

Ekstensi autentikasiGoogle Cloud

Runtime Beam mencakup ekstensi autentikasi Google Cloud yang otomatis mengautentikasi permintaan OpenTelemetry Protocol (OTLP) yang dikirim ke Google API seperti telemetry.googleapis.com.

Saat menggunakan default: Google Cloud autentikasi diaktifkan secara otomatis.
Saat mengekspor ke endpoint non-Google: Jika enable_otel_defaults tidak disertakan, Beam akan menonaktifkan ekstensi autentikasi Google Cloud dengan menyetel properti sistem google.otel.auth.target.signals=none.
Konfigurasi manual: Anda dapat menyetel properti sistem google.otel.auth.target.signals ke none (dinonaktifkan) atau trace (diaktifkan hanya untuk rekaman aktivitas).

Menulis rekaman aktivitas kustom di DoFn

Anda dapat membuat rentang kustom di dalam kode pengguna, seperti di dalam DoFn, untuk mengukur operasi tertentu. Hal ini mencakup panggilan ke API eksternal atau logika yang intensif secara komputasi.

Contoh penerapan DoFn

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.Scope;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.SdkHarnessOptions;
import org.apache.beam.sdk.transforms.DoFn;

public class InvestigateTransactionDoFn extends DoFn<Transaction, RiskAssessment> {
  private transient Tracer tracer;

  @Setup
  public void setup(PipelineOptions options) {
    // Obtain the Tracer instance from SdkHarnessOptions
    this.tracer = options
        .as(SdkHarnessOptions.class)
        .getOpenTelemetry()
        .getTracer("org.apache.beam.examples.adk.aml");
  }

  @ProcessElement
  public void processElement(@Element Transaction tx, OutputReceiver<RiskAssessment> out) {
    // Create and start a custom span as a child of the active context
    Span span = tracer.spanBuilder("InvestigateTransactionDoFn:Analyze")
        .setAttribute("transaction.id", tx.getId())
        .setAttribute("transaction.amount", tx.getAmount())
        .startSpan();

    // Make the span current in the execution thread
    try (Scope scope = span.makeCurrent()) {
      // Perform your logic (such as calling an external service)
      RiskAssessment assessment = analyzeTransactionWithLLM(tx);
      out.output(assessment);
    } catch (Exception e) {
      // Record exception details and set error status on the span
      span.recordException(e);
      span.setStatus(StatusCode.ERROR, e.getMessage());
      throw e;
    } finally {
      // Always end the span
      span.end();
    }
  }

  private RiskAssessment analyzeTransactionWithLLM(Transaction tx) {
    // Application analysis logic
    return new RiskAssessment();
  }
}

Mengintegrasikan dengan library klien

Saat DoFn memanggil library eksternal yang juga mendukung OpenTelemetry, seperti library klien Google Cloud , konfigurasi klien untuk menggunakan instance OpenTelemetry pipeline sehingga rentang turunan ditautkan langsung ke rekaman aktivitas pipeline.

Contoh klien Java Spanner

Untuk menjalankan kueri Spanner manual di dalam DoFn dan menautkan rentang tersebut ke rekaman aktivitas pipeline yang aktif, lakukan inisialisasi SpannerOptions dengan instance OpenTelemetry pipeline:

import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.DatabaseId;
import com.google.cloud.spanner.ResultSet;
import com.google.cloud.spanner.Spanner;
import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.spanner.Statement;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.SdkHarnessOptions;
import org.apache.beam.sdk.transforms.DoFn;

public class SpannerEnrichmentDoFn extends DoFn<String, String> {
  private transient Spanner spanner;
  private transient DatabaseClient dbClient;
  private final String instanceId;
  private final String databaseId;

  public SpannerEnrichmentDoFn(String instanceId, String databaseId) {
    this.instanceId = instanceId;
    this.databaseId = databaseId;
  }

  @Setup
  public void setup(PipelineOptions pipelineOptions) {
    // Build SpannerOptions leveraging the pipeline's active OpenTelemetry instance
    SpannerOptions options =
        SpannerOptions.newBuilder()
            .setOpenTelemetry(pipelineOptions
                .as(SdkHarnessOptions.class)
                .getOpenTelemetry())
            .setEnableEndToEndTracing(true)
            .setEnableExtendedTracing(true)
            .build();

    this.spanner = options.getService();
    DatabaseId db = DatabaseId.of(options.getProjectId(), instanceId, databaseId);
    this.dbClient = spanner.getDatabaseClient(db);
  }

  @ProcessElement
  public void processElement(@Element String inputId, OutputReceiver<String> out) {
    Statement statement = Statement.newBuilder(
        "SELECT Details FROM AccountTable WHERE AccountId = @id")
        .bind("id").to(inputId)
        .build();

    // Spanner queries automatically generate child spans linked to the active trace
    try (ResultSet rs = dbClient.singleUse().executeQuery(statement)) {
      while (rs.next()) {
        out.output(rs.getString("Details"));
      }
    }
  }

  @Teardown
  public void tearDown() {
    if (spanner != null) {
      spanner.close();
    }
  }
}

Penerapan konteks di seluruh tahapan

Dalam pipeline data terdistribusi, membawa konteks rekaman aktivitas di seluruh batas pekerja dan operasi pengacakan sangat penting untuk visibilitas menyeluruh.

Saat Anda memasukkan Redistribute.arbitrarily() ke dalam pipeline, konteks trace aktif akan diserialisasi ke dalam metadata elemen, ditransmisikan melalui jaringan, dan dideserialisasi di worker VM penerima. Hal ini memastikan kelanjutan rekaman aktivitas yang tidak terputus di seluruh tahap.

Contoh berikut menggambarkan cara konteks rekaman aktivitas disebarkan di seluruh tahap pipeline:

Mengekstrak konteks: KafkaIO.read() membaca rekaman masuk dan mengekstrak konteks pelacakan dari header pesan.
Memproses dan membuat rentang: DoFn A memproses elemen dalam rentang rekaman aktivitas yang aktif.
Menyebarkan di seluruh pekerja: Redistribute.arbitrarily() menserialisasi konteks rekaman aktivitas aktif ke dalam metadata elemen dan mengacak elemen di seluruh jaringan.
Melanjutkan trace: DoFn B menerima elemen di worker VM, mendeserialisasi konteks trace, dan melanjutkan pemrosesan dengan Trace ID yang sama.

Melihat rekaman aktivitas di Cloud Trace

Setelah mengaktifkan pelacakan, lihat rekaman aktivitas pipeline di konsol Google Cloud :

Di konsol Google Cloud , buka halaman Trace explorer.

Buka Trace Explorer
Di kotak filter, filter menurut Service: SERVICE_NAME. SERVICE_NAME ini cocok dengan appName pipeline Anda atau otel.service.name yang dikonfigurasi.

Atau, telusuri ID Rekaman Aktivitas tertentu.
Pilih rekaman aktivitas dari daftar hasil untuk melihat linimasa detailnya.

Grafik linimasa menampilkan:
- Rentang pelari: Rentang yang dihasilkan oleh runtime Beam. Misalnya, PubSubIO.Read dan ProcessElement.
- Rentang kustom: Rentang yang dibuat di dalam penerapan DoFn Anda. Contoh, InvestigateTransactionDoFn:Analyze.
- Rentang library klien: Rentang turunan yang dihasilkan oleh library klien terintegrasi. Misalnya, kueri RPC Spanner.

Langkah berikutnya

Pelajari lebih lanjut cara menjelajahi dan menganalisis trace terdistribusi di dokumentasi Cloud Trace.
Lacak pesan dari penyerapan hingga pemrosesan menggunakan pelacakan OpenTelemetry Pub/Sub.
Pantau throughput pipeline, pemanfaatan CPU, dan detail eksekusi dengan antarmuka pemantauan Dataflow.
Lihat log operasional dan diagnosis error menggunakan log pipeline Dataflow.
Selidiki dan selesaikan masalah performa dengan Memecahkan masalah tugas yang lambat atau macet dan Mendeteksi dan mengatasi bottleneck pipeline.