스트림 관리

이 페이지에서는 Datastream API를 사용하여 다음 작업을 하는 방법을 알아봅니다.

  • 스트림 만들기
  • 스트림 및 스트림 객체에 대한 정보 가져오기
  • 스트림 시작, 일시중지, 다시 시작, 수정, 스트림 객체에 대한 백필 시작 및 중지로 스트림 업데이트
  • 영구적으로 실패한 스트림 복구
  • Oracle 스트림의 대용량 객체 스트리밍 사용 설정
  • 스트림 삭제

Datastream API는 두 가지 방법으로 사용될 수 있습니다. REST API 호출을 수행하거나 Google Cloud CLI(CLI)를 사용할 수 있습니다.

Google Cloud CLI를 사용해서 Datastream 스트림을 관리하는 방법에 대한 개요는 gcloud CLI Datastream 스트림을 참고하세요.

스트림 만들기

이 섹션에서는 소스에서 대상으로 데이터를 전송하는 데 사용되는 스트림을 만드는 방법을 알아봅니다. 다음 예는 포괄적이지는 않지만 Datastream의 특정 기능을 강조합니다. 특정 사용 사례를 해결하려면 다음 예시를 Datastream API 참고 문서와 함께 사용하세요.

이 섹션에서는 다음 사용 사례를 설명합니다.

예시 1: BigQuery로 특정 객체 스트리밍

이 예시에서는 다음을 수행하는 방법을 알아봅니다.

  • MySQL에서 BigQuery로 스트리밍
  • 스트림에 객체 집합 포함
  • 스트림의 쓰기 모드를 추가 전용으로 정의
  • 스트림에 포함된 모든 객체 백필

다음은 schema1의 모든 테이블과 schema2의 두 가지 특정 테이블인 tableA와 tableC 테이블을 가져오는 요청입니다. 이벤트가 BigQuery의 데이터 세트에 기록됩니다.

요청에 customerManagedEncryptionKey 매개변수가 포함되지 않으므로 CMEK 대신 Google Cloud 내부 키 관리 시스템을 사용해서 데이터를 암호화합니다.

이전 백필(또는 스냅샷) 수행과 연결된 backfillAll 매개변수는 빈 딕셔너리({})로 설정됩니다. 즉, Datastream이 스트림에 포함된 모든 테이블의 이전 데이터를 백필합니다.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlCdcStream
{
  "displayName": "MySQL CDC to BigQuery",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "schema1" },
          {
            "database": "schema2",
            "mysqlTables": [
              {
                "table": "tableA",
                "table": "tableC"
              }
            ]
          }
        ]
      },
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "dataFreshness": "900s"
    }
  },
  "backfillAll": {}
}

gcloud

gcloud를 사용해서 스트림을 만드는 방법은 Google Cloud SDK 문서를 참조하세요.

예시 2: PostgreSQL 소스를 사용하여 스트림에서 특정 객체 제외

이 예시에서는 다음을 수행하는 방법을 알아봅니다.

  • PostgreSQL에서 BigQuery로 스트리밍
  • 스트림에서 객체 제외
  • 백필에서 객체 제외

다음 코드는 소스 PostgreSQL 데이터베이스에서 BigQuery로 데이터를 전송하는 데 사용되는 스트림 만들기 요청을 보여줍니다. 소스 PostgreSQL 데이터베이스에서 스트림을 만들 때는 요청에 PostgreSQL과 관련된 두 가지 추가 필드를 지정해야 합니다.

  • replicationSlot: 복제 슬롯은 복제용으로 PostgreSQL 데이터베이스를 구성하기 위한 기본 요건입니다. 각 스트림에 대해 복제 슬롯을 만들어야 합니다.
  • publication: 게시는 변경사항을 복제하려는 테이블 그룹입니다. 스트림을 시작하려면 먼저 게시 이름이 데이터베이스에 있어야 합니다. 게시에는 최소한 스트림의 includeObjects 목록에 지정된 테이블이 포함되어야 합니다.

이전 백필(또는 스냅샷) 수행과 관련된 backfillAll 매개변수는 테이블 하나를 제외하도록 설정됩니다.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myPostgresStream
{
  "displayName": "PostgreSQL to BigQueryCloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/postgresCp",
    "postgresqlSourceConfig": {
      "replicationSlot": "replicationSlot1",
      "publication": "publicationA",
      "includeObjects": {
        "postgresqlSchemas": {
          "schema": "schema1"
        }
      },
      "excludeObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
        "postgresqlTables": [
          {
            "table": "tableA",
            "postgresqlColumns": [
              { "column": "column5" }
              ]
              }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "dataFreshness": "900s",
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
           "location": "us",
           "datasetIdPrefix": "prefix_"
        }
      }
    }
  },
  "backfillAll": {
    "postgresqlExcludedObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
            "postgresqlTables": [
              { "table": "tableA" }
            ]
          }
        ]