Cloud Storage Avro to Spanner 데이터 검증 템플릿

베타

이 기능에는 서비스별 약관의 일반 서비스 약관 섹션에 있는 'GA 이전 제품 또는 서비스 약관'이 적용됩니다. GA 이전 기능은 '있는 그대로' 제공되며 지원이 제한될 수 있습니다. 자세한 내용은 출시 단계 설명을 참조하세요.

Cloud Storage Avro to Spanner Data Validation 템플릿은 Cloud Storage와 Spanner에서 데이터를 읽고 이를 비교하여 이전의 정확성을 검증하는 일괄 파이프라인입니다.

파이프라인 요구사항

• 파이프라인을 실행하기 전에 Avro 파일의 Cloud Storage 디렉터리가 있어야 합니다. gcsOutputDirectory 매개변수를 사용하여 SourceDB to Cloud Spanner 파이프라인을 실행하여 이러한 Avro 파일을 생성합니다.
• 검증 결과를 위한 대상 BigQuery 데이터 세트는 파이프라인을 실행하기 전에 있어야 합니다.
• 파이프라인을 실행하기 전에 Spanner 테이블이 있어야 합니다.
• Spanner 테이블에 호환 가능한 스키마가 있어야 합니다 (직접 또는 스키마 매핑을 통해).
• SourceDB to Cloud Spanner 파이프라인에서 실행한 마이그레이션을 검증하려면 공유 구성 (예: sessionFilePath, schemaOverridesFilePath, transformationJarPath)에 동일한 값을 전달해야 합니다.

템플릿 매개변수

필수 매개변수

• gcsInputDirectory: 이 디렉터리는 소스에서 읽은 레코드의 AVRO 파일을 읽는 데 사용됩니다. 예를 들면 gs://your-bucket/your-path입니다.
• projectId: Cloud Spanner 프로젝트의 이름입니다.
• instanceId: 대상 Cloud Spanner 인스턴스입니다.
• databaseId: 대상 Cloud Spanner 데이터베이스입니다.
• bigQueryDataset: 검증 결과가 저장될 BigQuery 데이터 세트 ID입니다. 예를 들면 validation_report_dataset입니다.

선택적 파라미터

• spannerHost: 템플릿에서 호출할 Cloud Spanner 엔드포인트입니다. 예를 들면 https://batch-spanner.googleapis.com입니다. 기본값은 https://batch-spanner.googleapis.com입니다.
• spannerPriority: Cloud Spanner 호출의 요청 우선순위입니다. 값은 [HIGH,MEDIUM,LOW] 중 하나여야 합니다. 기본값은 HIGH입니다.
• sessionFilePath: Spanner 마이그레이션 도구의 매핑 정보가 포함된 Cloud Storage의 세션 파일 경로입니다. 기본값은 빈 값입니다.
• schemaOverridesFilePath: 소스에서 Spanner로 테이블 및 열 이름 재정의를 지정하는 파일입니다. 기본값은 빈 값입니다.
• tableOverrides: 소스에서 Spanner로의 테이블 이름 재정의입니다. 다음 형식으로 작성됩니다. [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}] 이 예에서는 Singers 테이블을 Vocalists에, Albums 테이블을 레코드에 매핑하는 것을 보여줍니다. 예를 들면 [{Singers, Vocalists}, {Albums, Records}]입니다. 기본값은 빈 값입니다.
• columnOverrides: 소스에서 Spanner로의 열 이름 재정의입니다. 다음 형식으로 작성됩니다. [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}] SourceTableName은 소스 및 spanner 쌍에서 동일하게 유지되어야 합니다. 테이블 이름을 재정의하려면 tableOverrides를 사용합니다. 이 예시에서는 Singers 및 Albums 테이블에서 SingerName을 TalentName에, AlbumName을 RecordName에 각각 매핑하는 것을 보여줍니다. 예를 들면 [{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]입니다. 기본값은 빈 값입니다.
• runId: 유효성 검사 실행의 고유 식별자입니다. 제공되지 않으면 Dataflow 작업 이름이 사용됩니다. 예를 들면 run_20230101_120000입니다.
• transformationJarPath: 레코드를 처리하기 위한 맞춤 변환 로직이 포함된 Cloud Storage의 맞춤 jar 위치입니다. 기본값은 빈 값입니다.
• transformationClassName: 커스텀 변환 로직이 있는 정규화된 클래스 이름입니다. transformationJarPath가 지정된 경우 필수 필드입니다. 기본값은 빈 값입니다.
• transformationCustomParameters: 커스텀 변환 클래스에 전달할 커스텀 매개변수가 포함된 문자열입니다. 기본값은 빈 값입니다.
• tables: 검증 실행에 포함할 소스 테이블의 쉼표로 구분된 목록입니다. 기본값은 빈 값입니다.
• tableConfigurationFilePath: 유효성을 검사할 소스 테이블의 JSON 목록이 포함된 GCS 파일 경로입니다. 구조가 {"tableNames": ["table1", "table2"]}인 JSON 파일이어야 합니다. 선택적으로 optionalConfigurations.<sourceTable>.spannerQuery는 테이블을 읽는 데 사용되는 Spanner 쿼리를 설정합니다(예: 선택된 shardIds로 제한). Spanner 이름으로 Spanner 전용 테이블에 키를 지정합니다. 쿼리 규칙은 모듈 README.md의 '샤드 하위 집합 검증' 섹션을 참고하세요. 기본값은 빈 값입니다.
• shardIds: 논리적 샤드 ID의 쉼표로 구분된 목록입니다 (sourcedb-to-spanner 템플릿에서 사용되는 ID로, Avro 출력에서 <table>/<shardId>/ 디렉터리로 표시됨). 설정되면 해당 샤드만 검증됩니다. 세션 파일 ShardIdColumn이 있는 테이블은 해당 열에서 필터링됩니다. 범위 내의 다른 모든 테이블에는 tableConfigurationFilePath에 spannerQuery이 있어야 합니다. 그렇지 않으면 작업이 실패합니다. 예를 들면 shard_001,shard_007입니다. 기본값은 빈 값입니다.

템플릿 실행

콘솔

1. Dataflow 템플릿에서 작업 만들기 페이지로 이동합니다.
템플릿에서 작업 만들기로 이동
2. 작업 이름 필드에 고유한 작업 이름을 입력합니다.
3. (선택사항) 리전 엔드포인트의 드롭다운 메뉴에서 값을 선택합니다. 기본 리전은 us-central1입니다.

   Dataflow 작업을 실행할 수 있는 리전 목록은 Dataflow 위치를 참조하세요.

4. Dataflow 템플릿 드롭다운 메뉴에서 Cloud Storage Avro files to Spanner Data Validation 템플릿을 선택합니다.
5. 제공된 파라미터 필드에 파라미터 값을 입력합니다.
6. 작업 실행을 클릭합니다.

gcloud CLI

참고: Google Cloud CLI를 사용하여 flex템플릿을 실행하려면 Google Cloud CLI 버전 284.0.0 이상이 필요합니다.

셸 또는 터미널에서 템플릿을 실행합니다.

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/Avro_to_Spanner_Data_Validator \
    --region=REGION_NAME \
    --parameters \
       gcsInputDirectory=PATH_TO_AVRO_FILES,\
       projectId=PROJECT_ID,\
       instanceId=INSTANCE_ID,\
       databaseId=DATABASE_ID,\
       bigQueryDataset=BIGQUERY_DATASET

다음을 바꿉니다.

• JOB_NAME: 선택한 고유한 작업 이름
• VERSION: 사용할 템플릿 버전

  다음 값을 사용할 수 있습니다.

  • latest: 버킷의 날짜가 지정되지 않은 상위 폴더(gs://dataflow-templates/latest/)에서 사용할 수 있는 최신 버전의 템플릿을 사용합니다.
  • 버전 이름(예: 2023-09-12-00_RC00): 버킷의 날짜가 지정된 해당 상위 폴더(gs://dataflow-templates/)에 중첩되어 있는 특정 버전의 템플릿을 사용합니다.
  주의: 최신 버전의 템플릿이 브레이킹 체인지로 업데이트될 수 있습니다. 프로덕션 환경에서는 이러한 브레이킹 체인지가 프로덕션 워크플로에 영향을 미치지 않도록 최신 날짜가 지정된 상위 폴더에 보관된 템플릿을 사용해야 합니다.
• REGION_NAME: Dataflow 작업을 배포할 리전(예: us-central1)
• PATH_TO_AVRO_FILES: Avro 파일의 Cloud Storage 경로입니다.
• PROJECT_ID: Spanner 프로젝트 ID입니다.
• INSTANCE_ID: 대상 Spanner 인스턴스 ID입니다.
• DATABASE_ID: 대상 Spanner 데이터베이스 ID입니다.
• BIGQUERY_DATASET: 검증 결과의 BigQuery 데이터 세트 ID입니다.

API

REST API를 사용하여 템플릿을 실행하려면 HTTP POST 요청을 전송합니다. API 및 승인 범위에 대한 자세한 내용은 projects.templates.launch를 참조하세요.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "gcsInputDirectory": "PATH_TO_AVRO_FILES",
       "projectId": "PROJECT_ID",
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "bigQueryDataset": "BIGQUERY_DATASET"
     },
     "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Avro_to_Spanner_Data_Validator"
  }
}

다음을 바꿉니다.

• PROJECT_ID: Dataflow 작업을 실행하려는 Google Cloud 프로젝트 ID
• JOB_NAME: 선택한 고유한 작업 이름
• VERSION: 사용할 템플릿 버전

  다음 값을 사용할 수 있습니다.

  • latest: 버킷의 날짜가 지정되지 않은 상위 폴더(gs://dataflow-templates/latest/)에서 사용할 수 있는 최신 버전의 템플릿을 사용합니다.
  • 버전 이름(예: 2023-09-12-00_RC00): 버킷의 날짜가 지정된 해당 상위 폴더(gs://dataflow-templates/)에 중첩되어 있는 특정 버전의 템플릿을 사용합니다.
  주의: 최신 버전의 템플릿이 브레이킹 체인지로 업데이트될 수 있습니다. 프로덕션 환경에서는 이러한 브레이킹 체인지가 프로덕션 워크플로에 영향을 미치지 않도록 최신 날짜가 지정된 상위 폴더에 보관된 템플릿을 사용해야 합니다.
• LOCATION: Dataflow 작업을 배포할 리전(예: us-central1)
• PATH_TO_AVRO_FILES: Avro 파일의 Cloud Storage 경로입니다.
• PROJECT_ID: Spanner 프로젝트 ID입니다.
• INSTANCE_ID: 대상 Spanner 인스턴스 ID입니다.
• DATABASE_ID: 대상 Spanner 데이터베이스 ID입니다.
• BIGQUERY_DATASET: 검증 결과의 BigQuery 데이터 세트 ID입니다.

템플릿 소스 코드

Java

/*
 * Copyright (C) 2026 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates;

import static com.google.cloud.teleport.v2.constants.GCSSpannerDVConstants.SOURCE_TAG;
import static com.google.cloud.teleport.v2.constants.GCSSpannerDVConstants.SPANNER_TAG;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.config.TableConfiguration;
import com.google.cloud.teleport.v2.dto.ComparisonRecord;
import com.google.cloud.teleport.v2.fn.SchemaMapperProviderFn;
import com.google.cloud.teleport.v2.options.GCSSpannerDVOptions;
import com.google.cloud.teleport.v2.spanner.ddl.Ddl;
import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper;
import com.google.cloud.teleport.v2.spanner.migrations.transformation.CustomTransformation;
import com.google.cloud.teleport.v2.transforms.MatchRecordsTransform;
import com.google.cloud.teleport.v2.transforms.ReportResultsTransform;
import com.google.cloud.teleport.v2.transforms.SourceReaderTransform;
import com.google.cloud.teleport.v2.transforms.SpannerInformationSchemaProcessorTransform;
import com.google.cloud.teleport.v2.transforms.SpannerReaderTransform;
import com.google.common.annotations.VisibleForTesting;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.PCollectionView;
import org.joda.time.Instant;

@Template(
    name = "Avro_to_Spanner_Data_Validator",
    category = TemplateCategory.BATCH,
    displayName = "Cloud Storage Avro files to Spanner Data Validation",
    description =
        "Batch pipeline that reads data from Cloud Storage and Spanner and compares them to validate migration"
            + " correctness.",
    optionsClass = GCSSpannerDVOptions.class,
    flexContainerName = "avro-to-spanner-dv",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/avro-to-spanner-dv",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The Cloud Storage directory for Avro files must exist before pipeline execution.",
      "The target BigQuery dataset for validation results must exist before pipeline execution.",
      "The Spanner tables must exist before pipeline execution.",
      "The Spanner tables must have a compatible schema (either directly or through schema mapping)."
    })
public class GCSSpannerDV {

  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    GCSSpannerDVOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(GCSSpannerDVOptions.class);
    run(options);
  }

  public static PipelineResult run(GCSSpannerDVOptions options) {
    Pipeline pipeline = Pipeline.create(options);

    TableConfiguration tableConfig = TableConfiguration.parseFromOptions(options);

    SpannerConfig spannerConfig = createSpannerConfig(options);

    // Fetch Spanner DDL using Info schema
    final PCollectionView<Ddl> ddlView =
        pipeline.apply(
            "ReadSpannerInformationSchema",
            new SpannerInformationSchemaProcessorTransform(spannerConfig));

    // Get Schema mapper provider, we get Ddl from a side input
    // so the mapper has to be initialized lazily
    SerializableFunction<Ddl, ISchemaMapper> schemaMapperProvider =
        new SchemaMapperProviderFn(
            options.getSessionFilePath(),
            options.getSchemaOverridesFilePath(),
            options.getTableOverrides(),
            options.getColumnOverrides());

    CustomTransformation customTransformation =
        CustomTransformation.builder(
                options.getTransformationJarPath(), options.getTransformationClassName())
            .setCustomParameters(options.getTransformationCustomParameters())
            .build();

    // Get Source records hashes
    PCollection<ComparisonRecord> sourceRecords =
        pipeline.apply(
            "ReadSourceRecords",
            new SourceReaderTransform(
                options.getGcsInputDirectory(),
                ddlView,
                schemaMapperProvider,
                customTransformation,
                tableConfig));

    // Get Spanner records hashes
    PCollection<ComparisonRecord> spannerRecords =
        pipeline.apply(
            "ReadSpannerRecords",
            new SpannerReaderTransform(spannerConfig, ddlView, schemaMapperProvider, tableConfig));

    PCollectionTuple inputs =
        PCollectionTuple.of(SOURCE_TAG, sourceRecords).and(SPANNER_TAG, spannerRecords);

    // Match records to determine equivalence
    PCollectionTuple matchResults = inputs.apply("MatchRecords", new MatchRecordsTransform());

    // Report results of the validation
    Instant startTimestamp = Instant.now();
    String runId = options.getRunId();
    if (runId == null) {
      runId = String.format("%s_%s", options.getJobName(), startTimestamp);
    }

    matchResults.apply(
        "ReportResults",
        new ReportResultsTransform(options.getBigQueryDataset(), runId, startTimestamp));

    return pipeline.run();
  }

  @VisibleForTesting
  static SpannerConfig createSpannerConfig(GCSSpannerDVOptions options) {
    return SpannerConfig.create()
        .withProjectId(ValueProvider.StaticValueProvider.of(options.getProjectId()))
        .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
        .withInstanceId(ValueProvider.StaticValueProvider.of(options.getInstanceId()))
        .withDatabaseId(ValueProvider.StaticValueProvider.of(options.getDatabaseId()))
        .withRpcPriority(ValueProvider.StaticValueProvider.of(options.getSpannerPriority()));
  }
}