Google은 AI 기술을 사용하여 콘텐츠를 사용자의 기본 언어로 번역합니다. AI 번역에는 오류가 있을 수 있습니다.

Dataflow에서 OpenTelemetry 분산 추적 사용

분산 스트리밍 파이프라인에서는 집계된 측정항목만으로는 지연 시간 급증의 원인이 되는 특정 단계, 외부 서비스 호출 또는 작업자 셔플을 식별하기 어려울 수 있습니다. Dataflow에서 OpenTelemetry 분산 추적을 사용하면 단일 추적 ID로 파이프라인 변환, 네트워크 경계, 통합 서비스 전반에서 개별 요소를 엔드 투 엔드로 추적할 수 있습니다. 이 가시성을 통해 성능 병목 현상을 정확히 파악하고, 파이프라인 실행 비용을 최적화하고, 복잡한 스트리밍 워크로드를 해결할 수 있습니다.

다음 정보를 사용하여 Apache Beam 파이프라인에서 OpenTelemetry 분산 추적을 사용 설정하고, 구성하고, 사용할 수 있습니다.

사용 사례

Dataflow의 분산 추적은 다음과 같은 시나리오에서 특히 유용합니다.

스트리밍 워크로드: 요소별 처리 시간을 추적하고 지연 또는 지연 시간 급증을 유발하는 단계를 식별합니다.
사용자 코드의 외부 서비스 호출 추적: 파이프라인이 사용자 코드 (예: DoFn) 내에서 외부 데이터베이스, 마이크로서비스 또는 생성형 AI API를 호출할 때 실행 시간을 측정합니다. 예를 들어 외부 데이터베이스 (예: Spanner 또는 Bigtable), 마이크로서비스 (HTTP 또는 gRPC 사용) 또는 생성형 AI API (예: Gemini)에 대한 호출 시간을 측정할 수 있습니다. 자세한 내용은 DoFn에서 맞춤 트레이스 작성 및 클라이언트 라이브러리와 통합을 참고하세요.
비용 및 성능 최적화: 실제 부하에서 PTransform 인스턴스의 특정 작업 기간을 노출하여 비효율적인 직렬화, 느린 쿼리 또는 작업자 스레드 경합을 식별합니다.
복잡한 에이전트 및 워크플로 파이프라인: 여러 브랜치, 반복적인 에이전트 루프, 조인이 있는 파이프라인을 파악하여 가장 오래 걸리는 변환 또는 브랜치를 확인할 수 있습니다.

제한사항

Dataflow의 분산 추적에는 다음과 같은 제한사항이 있습니다.

이 기능은 스트리밍 파이프라인 전용으로 설계되고 검증되었습니다.
이 기능을 사용하려면 Java용 Beam SDK(버전 2.76.0 이상)를 사용해야 합니다.
교차 스테이지 컨텍스트 전파에는 스트리밍 Java Runner(이전 명칭: Runner v1)가 필요합니다. 포터블 러너 (이전 명칭: 러너 v2)는 셔플 경계에서 트레이스 컨텍스트를 재설정합니다.
Beam I/O 커넥터의 내장 OpenTelemetry 추적은 Pub/Sub, Apache Kafka, Spanner 변경 내역 커넥터에서만 지원됩니다. Apache Kafka의 경우 KafkaIO 커넥터를 직접 사용하는 경우에만 추적이 지원됩니다. Kafka용 관리형 I/O를 사용하는 경우에는 지원되지 않습니다. 다른 I/O 커넥터 (예: BigtableIO)에는 기본 제공 커넥터 추적이 없습니다. 이러한 서비스와의 작업을 추적하려면 사용자 코드 (예: DoFn) 내에서 호출을 계측하세요.
여러 요소를 하나로 결합하는 작업은 단일 트레이스 컨텍스트를 전파하지 않습니다. 여기에는 GroupByKey, CoGroupByKey, Combine와 같은 작업이 포함됩니다.
타이머 설정 (Timer.set())과 @OnTimer 콜백 메서드 실행 간에 컨텍스트가 전파되지 않습니다.

기본 요건

Dataflow에서 분산 추적을 사용하려면 환경이 다음 요구사항을 충족해야 합니다.

작업은 다음 요구사항을 충족해야 합니다.
- 스트리밍 파이프라인입니다.
- 스트리밍 Java 러너(이전 명칭: 러너 v1)를 사용합니다.
- Apache Beam Java SDK 버전 2.76.0 이상을 사용합니다.
Cloud Trace로 트레이스를 내보내는 경우 Google Cloud 프로젝트에서 Telemetry API(telemetry.googleapis.com)를 사용 설정합니다.

API 사용 설정에 필요한 역할
API를 사용 설정하려면 serviceusage.services.enable 권한이 필요합니다. 프로젝트를 만든 경우 소유자 역할 (roles/owner)을 통해 이 권한이 이미 있을 수 있습니다. 그렇지 않은 경우 서비스 사용량 관리자 역할 (roles/serviceusage.serviceUsageAdmin)을 통해 이 권한을 얻을 수 있습니다. 역할을 부여하는 방법 알아보기
API 사용 설정하기
작업자 서비스 계정에는 cloudtrace.traces.patch 권한이 포함된 Cloud Trace 에이전트 (roles/cloudtrace.agent) 역할이 있어야 합니다.

핵심 개념

분산 추적은 다음 OpenTelemetry 개념을 사용합니다.

Trace: 분산 시스템을 통해 흐르는 완전한 트랜잭션 또는 워크플로의 표현입니다. 트레이스는 하나 이상의 스팬으로 구성됩니다.
스팬: 단일 작업 단위를 나타내는 트레이스의 기본 구성요소입니다. 예를 들어 변환 실행, 메서드 호출 또는 RPC 요청이 있습니다.
컨텍스트: 스레드와 API 경계에 전파되는 상태입니다. 여기에는 추적 ID와 스팬 ID가 포함됩니다.
전파기: 네트워크 경계 전반에서 컨텍스트 표현식을 삽입하고 추출하는 데 사용되는 메커니즘입니다. 예를 들어 W3C Trace Context가 있습니다.
내보내기 도구: 수집된 스팬 데이터를 Cloud Trace 또는 자체 호스팅 OpenTelemetry Collector와 같은 추적 백엔드로 전송하는 구성요소
샘플러: 기록되고 내보내지는 트레이스를 결정하여 트레이스 볼륨을 제어하는 메커니즘입니다. 예를 들어 1% 샘플링 비율입니다.

OpenTelemetry 추적 사용 설정

기본 설정으로 OpenTelemetry 추적을 사용 설정하려면 작업을 제출할 때 필수 파이프라인 옵션과 실험을 구성하세요. 기본 설정은 1% 샘플링으로 Cloud Trace로 내보냅니다.

기본값을 수정하려면 추적 속성 설정 및 수정을 참고하세요.

작업을 실행할 때 파이프라인 옵션과 실험을 구성합니다.
비Flex 템플릿 작업

작업을 제출할 때 다음 실험 플래그를 전달합니다.
```
--experiments=enable_otel_defaults,element_metadata_supported,disable_portable_worker
```
Flex 템플릿 작업

gcloud 명령어에서 --additional-experiments 플래그를 사용하여 실험을 전달합니다.
```
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
```
다음을 바꿉니다.
- JOB_NAME: Dataflow 작업의 이름입니다.
- BUCKET_NAME: 템플릿이 포함된 Cloud Storage 버킷입니다.
- TEMPLATE_NAME: Flex 템플릿 파일의 이름입니다.
- REGION: 작업을 실행할 Google Cloud 리전입니다.
이러한 실험은 다음 동작을 구성합니다.
- enable_otel_defaults: Cloud Trace 내보내기 도구 및 1% 샘플링을 비롯한 표준 OpenTelemetry 기본값을 구성합니다.
- element_metadata_supported: 파이프라인 단계 간에 메타데이터 (예: 추적 컨텍스트)의 직렬화 및 전파를 사용 설정합니다.
- disable_portable_worker: Dataflow 스트리밍 Java Runner (이전 명칭: Runner v1)의 사용을 강제합니다.

Java용 Beam SDK 버전 2.76.0 또는 2.77.0을 사용하는 경우 빌드 구성에 필수 OpenTelemetry 종속 항목을 추가합니다.

버전 2.76.0 및 2.77.0의 문제로 인해 트레이스를 내보내려면 pom.xml 파일에 다음 종속 항목을 추가해야 합니다. 자세한 내용은 파이프라인 종속 항목 관리를 참고하세요.

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

파이프라인 코드에서 지원되는 I/O 커넥터에 OpenTelemetry 추적을 명시적으로 사용 설정합니다.
Pub/Sub

Pub/Sub에서 읽거나 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"));
```
다음을 바꿉니다.
- PROJECT_ID: Google Cloud 프로젝트의 ID입니다.
- SUBSCRIPTION_ID: Pub/Sub 구독의 이름입니다.
- TOPIC_ID: Pub/Sub 주제의 이름
컨텍스트 전파 세부정보:
- 업스트림 게시자: 외부 애플리케이션에서 메시지를 게시할 때 W3C 추적 컨텍스트를 Pub/Sub 메시지 속성에 삽입합니다. 자세한 내용은 Pub/Sub OpenTelemetry 추적을 참고하세요.
- 속성: Beam은 다음 속성을 예상하고 채웁니다.
  - googclient_traceparent: W3C traceparent 헤더를 전달합니다.
  - googclient_tracestate: W3C tracestate 헤더를 전달합니다.
- 헤더 형식은 W3C Trace 컨텍스트 사양을 참고하세요.
Apache Kafka

참고: Kafka 추적은 KafkaIO 커넥터를 직접 사용하는 경우에만 지원됩니다. Kafka용 관리형 I/O를 사용하는 경우에는 지원되지 않습니다.

Apache Kafka에서 읽거나 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
        );
```
컨텍스트 전파 세부정보:
- KafkaIO는 표준 Kafka 레코드 헤더를 사용하여 컨텍스트를 전파합니다.
- W3C Trace 컨텍스트 사양을 준수하는 헤더를 예상하고 삽입합니다.
  - traceparent: 수신되는 추적 컨텍스트를 식별합니다.
  - tracestate: 추가 공급업체별 라우팅 및 필터링 메타데이터를 제공합니다.
Spanner

Spanner 변경 내역에서 읽을 때 트레이스를 사용 설정하려면 다음 단계를 따르세요.
```
pipeline.apply("ReadChangeStream",
    SpannerIO.readChangeStream()
        .withSpannerConfig(spannerConfig)
        .withEnableOpenTelemetryTracing(true)
        .withChangeStreamName("CHANGE_STREAM_NAME"));
```
다음을 바꿉니다.
- CHANGE_STREAM_NAME: Spanner 변경 내역의 이름입니다.
스팬 생성 세부정보:
- 추적을 사용 설정하면 변경 데이터 캡처 (CDC) 리더가 읽은 모든 데이터베이스 변경 레코드에 대해 새 트레이스와 스팬이 시작됩니다.
- 새 레코드를 반환하지 않는 Spanner 리더 폴링 주기에서도 Cloud Trace UI에서 필터링할 수 있는 추적 스팬이 생성될 수 있습니다.

추적 속성 설정 및 수정

enable_otel_defaults을 지정하면 Dataflow는 다음 기본 속성을 적용합니다.

속성	기본값	설명
`otel.traces.exporter`	`otlp`	OpenTelemetry 프로토콜 (OTLP)을 사용하여 trace를 내보냅니다.
`otel.exporter.otlp.endpoint`	`https://telemetry.googleapis.com`	Cloud Trace OTLP 수신자를 타겟팅합니다.
`google.cloud.project`	프로젝트 ID	추적이 저장된 Google Cloud 프로젝트입니다.
`otel.traces.sampler.arg`	`0.01` (1%)	100개의 트레이스 중 1개를 기록합니다.
`otel.service.name`	`options.getAppName()`	파이프라인 `appName` 옵션에서 파생된 스팬과 연결된 서비스 이름입니다.

OpenTelemetry 구성을 맞춤설정하려면 세미콜론으로 구분된 속성 목록을 --openTelemetryProperties 파이프라인 옵션에 전달하세요.

다음 예에서는 모든 기본 속성을 명시적으로 구성합니다.

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

맞춤 서비스 이름 설정

기본적으로 추적 서비스 이름은 파이프라인 appName 옵션(--appName=YourPipelineName)에서 파생됩니다. otel.service.name 속성을 설정하여 이 값을 재정의할 수 있습니다.

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

샘플링 레이트 및 속성 변경

샘플링 레이트를 높게 설정하면 더 강력한 추적 정보를 얻을 수 있습니다. 하지만 높은 샘플링 비율 (예: 1.0 또는 always_on)은 Cloud Trace로 전송되는 추적 데이터의 양을 크게 늘려 Google Cloud 비용을 증가시킬 수 있습니다. 단기 디버깅에만 높은 샘플링 레이트를 사용하고 프로덕션 워크로드에서는 샘플링 레이트를 줄입니다.

다음 예에서는 샘플링 레이트를 명시적으로 구성합니다. 이 경우 인수의 전체 목록을 제공해야 합니다.

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

서드 파티 백엔드로 내보내기

Cloud Trace 대신 외부 관리형 추적 백엔드 또는 자체 호스팅 OpenTelemetry Collector로 trace를 전송하려면 다음 단계를 따르세요.

다음 실험은 생략합니다. --experiments=enable_otel_defaults

--openTelemetryProperties를 사용하여 내보내기 도구 속성을 수동으로 구성합니다.

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

Google Cloud 인증 확장 프로그램

Beam 런타임에는 telemetry.googleapis.com와 같은 Google API로 전송된 OpenTelemetry 프로토콜 (OTLP) 요청을 자동으로 인증하는 Google Cloud 인증 확장 프로그램이 포함되어 있습니다.

기본값 사용 시: Google Cloud 인증이 자동으로 사용 설정됩니다.
Google 이외의 엔드포인트로 내보낼 때: enable_otel_defaults가 생략되면 Beam은 시스템 속성 google.otel.auth.target.signals=none를 설정하여 Google Cloud 인증 확장 프로그램을 사용 중지합니다.
수동 구성: google.otel.auth.target.signals 시스템 속성을 none (사용 중지됨) 또는 trace (트레이스에만 사용 설정됨)로 설정할 수 있습니다.

DoFn에서 맞춤 트레이스 작성

DoFn와 같은 사용자 코드 내에서 맞춤 스팬을 만들어 특정 작업을 측정할 수 있습니다. 여기에는 외부 API 호출 또는 컴퓨팅 집약적 로직이 포함됩니다.

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();
  }
}

클라이언트 라이브러리와 통합

DoFn가 OpenTelemetry도 지원하는 외부 라이브러리(예: Google Cloud 클라이언트 라이브러리)를 호출하는 경우 하위 스팬이 파이프라인 추적에 직접 연결되도록 파이프라인의 OpenTelemetry 인스턴스를 사용하도록 클라이언트를 구성합니다.

Spanner Java 클라이언트 예시

DoFn 내에서 수동 Spanner 쿼리를 실행하고 이러한 스팬을 활성 파이프라인 트레이스에 연결하려면 파이프라인의 OpenTelemetry 인스턴스로 SpannerOptions을 초기화합니다.

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();
    }
  }
}

단계 간 컨텍스트 전파

분산 데이터 파이프라인에서는 작업자 경계와 셔플 작업 전반에 걸쳐 트레이스 컨텍스트를 전달하는 것이 엔드 투 엔드 가시성에 필수적입니다.

파이프라인에 Redistribute.arbitrarily()를 삽입하면 활성 추적 컨텍스트가 요소 메타데이터로 직렬화되어 네트워크를 통해 전송되고 수신 작업자 VM에서 역직렬화됩니다. 이렇게 하면 단계 간에 추적이 끊어지지 않고 계속됩니다.

다음 예시에서는 추적 컨텍스트가 파이프라인 단계에서 전파되는 방법을 보여줍니다.

컨텍스트 추출: KafkaIO.read()는 수신 레코드를 읽고 메시지 헤더에서 트레이스 컨텍스트를 추출합니다.
스팬 처리 및 생성: DoFn A는 활성 추적 스팬 내에서 요소를 처리합니다.
작업자 간 전파: Redistribute.arbitrarily()는 활성 트레이스 컨텍스트를 요소 메타데이터로 직렬화하고 네트워크 전반에서 요소를 셔플합니다.
트레이스 계속: DoFn B는 작업자 VM에서 요소를 수신하고, 트레이스 컨텍스트를 역직렬화하고, 동일한 트레이스 ID로 처리를 계속합니다.

Cloud Trace에서 trace 보기

트레이싱을 사용 설정한 후 Google Cloud 콘솔에서 파이프라인 트레이스를 확인합니다.

Google Cloud 콘솔에서 Trace 탐색기 페이지로 이동합니다.

Trace 탐색기로 이동
필터 상자에서 Service: SERVICE_NAME로 필터링합니다. 이 SERVICE_NAME는 파이프라인 appName 또는 구성된 otel.service.name와 일치합니다.

또는 특정 Trace ID를 검색합니다.
결과 목록에서 트레이스를 선택하여 자세한 타임라인을 확인합니다.

타임라인 그래프에는 다음이 표시됩니다.
- 러너 스팬: Beam 런타임에서 생성된 스팬입니다. 예를 들면 PubSubIO.Read 및 ProcessElement입니다.
- 맞춤 스팬: DoFn 구현 내에서 생성된 스팬입니다. 예를 들면 InvestigateTransactionDoFn:Analyze입니다.
- 클라이언트 라이브러리 스팬: 통합된 클라이언트 라이브러리에 의해 생성된 하위 스팬입니다. 예를 들어 Spanner RPC 쿼리가 있습니다.

다음 단계

Cloud Trace 문서에서 분산 trace 탐색 및 분석에 대해 자세히 알아보세요.
Pub/Sub OpenTelemetry 추적을 사용하여 수집부터 처리까지 메시지를 추적합니다.
Dataflow 모니터링 인터페이스를 사용하여 파이프라인 처리량, CPU 사용률, 실행 세부정보를 모니터링합니다.
Dataflow 파이프라인 로그를 사용하여 작업 로그를 확인하고 오류를 진단합니다.
느리거나 중단된 작업 문제 해결 및 파이프라인 병목 현상 감지 및 해결을 사용하여 성능 문제를 조사하고 해결합니다.