建立遷移工作流程

在 BigQuery Migration API 中建立遷移工作流程。遷移工作流程包含一系列子工作，執行這些子工作即可將資料倉儲遷移至 BigQuery。

程式碼範例

Java

在試用這個範例之前，請先按照「使用用戶端程式庫的 BigQuery 快速入門導覽課程」中的 Java 設定操作說明進行操作。詳情請參閱 BigQuery Java API 參考文件。

如要向 BigQuery 進行驗證，請設定應用程式預設憑證。詳情請參閱「設定用戶端程式庫的驗證機制」。

import com.google.api.gax.rpc.AlreadyExistsException;
import com.google.api.gax.rpc.ApiException;
import com.google.api.gax.rpc.InvalidArgumentException;
import com.google.cloud.bigquery.migration.v2.BigQueryDialect;
import com.google.cloud.bigquery.migration.v2.CreateMigrationWorkflowRequest;
import com.google.cloud.bigquery.migration.v2.Dialect;
import com.google.cloud.bigquery.migration.v2.LocationName;
import com.google.cloud.bigquery.migration.v2.MigrationServiceClient;
import com.google.cloud.bigquery.migration.v2.MigrationTask;
import com.google.cloud.bigquery.migration.v2.MigrationWorkflow;
import com.google.cloud.bigquery.migration.v2.TeradataDialect;
import com.google.cloud.bigquery.migration.v2.TranslationConfigDetails;
import java.io.IOException;
import java.util.Map;

public class CreateMigrationWorkflow {

  /**
   * Creates a Teradata SQL translation migration workflow to batch translate Teradata SQL scripts
   * and DDL into BigQuery-compatible SQL. The translation task reads input files from a Cloud
   * Storage source path and writes the converted output to a Cloud Storage target path.
   *
   * @param projectId The Google Cloud project ID (for example, "example-project-id").
   * @param location The geographic location of the migration workflow (for example, "us").
   * @param gcsSourcePath The Cloud Storage path for a directory of files to translate in a batch
   *     (for example, "gs://example-bucket/example-input-folder/").
   * @param gcsTargetPath The Cloud Storage path to write back the translated output files to (for
   *     example, "gs://example-bucket/example-output-folder/").
   */
  public static void createMigrationWorkflow(
      String projectId, String location, String gcsSourcePath, String gcsTargetPath) {
    LocationName parent = LocationName.of(projectId, location);

    Dialect sourceDialect =
        Dialect.newBuilder()
            .setTeradataDialect(
                TeradataDialect.newBuilder().setMode(TeradataDialect.Mode.SQL).build())
            .build();

    Dialect targetDialect =
        Dialect.newBuilder().setBigqueryDialect(BigQueryDialect.getDefaultInstance()).build();

    TranslationConfigDetails translationConfigDetails =
        TranslationConfigDetails.newBuilder()
            .setGcsSourcePath(gcsSourcePath)
            .setGcsTargetPath(gcsTargetPath)
            .setSourceDialect(sourceDialect)
            .setTargetDialect(targetDialect)
            .build();

    MigrationTask task =
        MigrationTask.newBuilder()
            .setType("Translation_Teradata2BQ")
            .setTranslationConfigDetails(translationConfigDetails)
            .build();

    MigrationWorkflow migrationWorkflow =
        MigrationWorkflow.newBuilder()
            .setDisplayName("Example Teradata to BigQuery Migration Workflow")
            .putTasks("translation-task", task)
            .build();

    CreateMigrationWorkflowRequest request =
        CreateMigrationWorkflowRequest.newBuilder()
            .setParent(parent.toString())
            .setMigrationWorkflow(migrationWorkflow)
            .build();

    try (MigrationServiceClient client = MigrationServiceClient.create()) {
      MigrationWorkflow response = client.createMigrationWorkflow(request);
      System.out.printf("Created migration workflow: %s%n", response.getName());
      System.out.printf("  Display Name: %s%n", response.getDisplayName());
      System.out.printf("  State: %s%n", response.getState());
      System.out.println("  Tasks:");
      for (Map.Entry<String, MigrationTask> entry : response.getTasksMap().entrySet()) {
        MigrationTask createdTask = entry.getValue();
        System.out.printf("    - %s:%n", entry.getKey());
        System.out.printf("        Type: %s%n", createdTask.getType());
        System.out.printf("        State: %s%n", createdTask.getState());
      }
    } catch (AlreadyExistsException err) {
      System.err.printf(
          "error: migration workflow already exists in location '%s' for project '%s': %s%n",
          location, projectId, err.getMessage());
    } catch (InvalidArgumentException err) {
      System.err.printf(
          "error: invalid argument creating migration workflow in location '%s' for project '%s': %s%n",
          location, projectId, err.getMessage());
    } catch (ApiException | IOException err) {
      System.err.printf(
          "error: failed to create migration workflow in location '%s' for project '%s': %s%n",
          location, projectId, err.getMessage());
    }
  }

Node.js

在試用這個範例之前，請先按照「使用用戶端程式庫的 BigQuery 快速入門導覽課程」中的 Node.js 設定操作說明進行操作。詳情請參閱 BigQuery Node.js API 參考文件。

如要向 BigQuery 進行驗證，請設定應用程式預設憑證。詳情請參閱「設定用戶端程式庫的驗證機制」。

const {MigrationServiceClient} = require('@google-cloud/bigquery-migration').v2;
const {status} = require('@grpc/grpc-js');

const client = new MigrationServiceClient();

/**
 * Creates a migration workflow.
 *
 * Batch translate Teradata SQL scripts and DDL into BigQuery-compatible SQL.
 * It configures a translation task that reads input files from a
 * Google Cloud Storage source bucket and writes the converted output to a target bucket.
 *
 * @param {string} projectId The Google Cloud project ID (for example, 'example-project-id').
 * @param {string} location The Google Cloud location (for example, 'us').
 * @param {string} gcsSourcePath The Cloud Storage path for the source data (for example, 'gs://example-bucket/example-input-folder').
 * @param {string} gcsTargetPath The Cloud Storage path for the translated results (for example, 'gs://example-bucket/example-output-folder').
 */
async function createMigrationWorkflow(
  projectId,
  location,
  gcsSourcePath,
  gcsTargetPath,
) {
  const parent = client.locationPath(projectId, location);
  const migrationWorkflow = {
    displayName: 'Example BTEQ Migration Workflow',
    tasks: {
      'translation-task': {
        type: 'Translation_Teradata2BQ',
        translationConfigDetails: {
          gcsSourcePath,
          gcsTargetPath,
          sourceDialect: {
            teradataDialect: {
              mode: 'SQL',
            },
          },
          targetDialect: {
            bigqueryDialect: {},
          },
        },
      },
    },
  };

  const request = {
    parent,
    migrationWorkflow,
  };

  try {
    const [workflow] = await client.createMigrationWorkflow(request);
    console.log(`Created migration workflow: ${workflow.name}`);
    console.log(`  Display Name: ${workflow.displayName}`);
    console.log(`  State: ${workflow.state}`);
    console.log('  Tasks:');
    for (const taskName in workflow.tasks) {
      const task = workflow.tasks[taskName];
      console.log(`    - ${taskName}:`);
      console.log(`        Type: ${task.type}`);
      console.log(`        State: ${task.state}`);
    }
  } catch (err) {
    if (err.code === status.INVALID_ARGUMENT) {
      console.error(
        `Error: Invalid argument provided for creating Migration '${migrationWorkflow.displayName}'. ` +
          `Details: ${err.message}. Please check the request parameters and ensure they are valid.`,
      );
    } else {
      console.error('Error creating migration workflow:', err);
    }
  }
}

Python

在試用這個範例之前，請先按照「使用用戶端程式庫的 BigQuery 快速入門導覽課程」中的 Python 設定操作說明進行操作。詳情請參閱 BigQuery Python API 參考文件。

如要向 BigQuery 進行驗證，請設定應用程式預設憑證。詳情請參閱「設定用戶端程式庫的驗證機制」。

from google.api_core import exceptions
from google.cloud import bigquery_migration_v2
from google.cloud.bigquery_migration_v2.types import (
    migration_entities,
    translation_config,
)


def create_migration_workflow(
    project_id: str, location: str, gcs_source_path: str, gcs_target_path: str
) -> None:
    """Creates a Teradata SQL translation workflow

    Creates a migration workflow to batch translate Teradata SQL scripts
    and DDL into BigQuery-compatible SQL. It configures a translation task
    that reads input files from a Google Cloud Storage source bucket and
    writes the converted output to a target bucket.

    Args:
        project_id: The Google Cloud project ID.
        location: The geographic location of the migration workflow (for example, us).
        gcs_source_path: The Cloud Storage path for a directory of files to
            translate in a batch (for example, gs://example-bucket/example-input-folder/).
        gcs_target_path: The Cloud Storage path to write back the corresponding
            input files to (for example, gs://example-bucket/example-output-folder/).
    """
    client = bigquery_migration_v2.MigrationServiceClient()

    parent = f"projects/{project_id}/locations/{location}"

    source_dialect = bigquery_migration_v2.Dialect()
    source_dialect.teradata_dialect = bigquery_migration_v2.TeradataDialect(
        mode=bigquery_migration_v2.TeradataDialect.Mode.SQL
    )
    target_dialect = bigquery_migration_v2.Dialect()
    target_dialect.bigquery_dialect = bigquery_migration_v2.BigQueryDialect()

    translation_config_details = bigquery_migration_v2.TranslationConfigDetails(
        gcs_source_path=gcs_source_path,
        gcs_target_path=gcs_target_path,
        source_dialect=source_dialect,
        target_dialect=target_dialect,
    )

    task = migration_entities.MigrationTask(
        type="Translation_Teradata2BQ",
        translation_config_details=translation_config_details,
    )

    workflow = migration_entities.MigrationWorkflow(
        display_name="Example Teradata to BigQuery Migration Workflow",
        tasks={"translation-task": task},
    )

    try:
        response = client.create_migration_workflow(
            parent=parent,
            migration_workflow=workflow,
        )
        print(f"Created migration workflow: {response.name}")
        print(f"Display name: {response.display_name}")
        print(f"State: {response.state.name}")
    except exceptions.AlreadyExists as e:
        print(f"Migration workflow already exists: {e}")

後續步驟

如要搜尋及篩選其他 Google Cloud 產品的程式碼範例，請參閱Google Cloud 範例瀏覽工具。