Эта страница переведена с помощью Cloud Translation API.

Создание сбора данных об изменениях с использованием Dataproc и Cloud Pub/Sub

1. Введение

Последнее обновление: 2025-06-19

Что такое сбор измененных данных?

Change Data Capture (CDC) — это набор шаблонов проектирования программного обеспечения, используемых для определения и отслеживания данных, которые изменились в базе данных. Проще говоря, это способ захвата и записи изменений, внесенных в данные, чтобы эти изменения можно было реплицировать в другие системы.

Сбор измененных данных (CDC) невероятно полезен в широком спектре сценариев, связанных с данными, таких как миграция данных, хранение и аналитика данных в реальном времени, аварийное восстановление и высокая доступность, аудит и соответствие требованиям и т. д.

Миграция данных

CDC упрощает проекты по миграции данных, позволяя осуществлять пошаговую передачу данных, сокращая время простоя и сводя к минимуму сбои.

Хранилище данных и аналитика в реальном времени

CDC обеспечивает постоянное обновление хранилищ данных и аналитических систем с учетом последних изменений из оперативных баз данных.

Это позволяет предприятиям принимать решения на основе информации в режиме реального времени.

Аварийное восстановление и высокая доступность

CDC позволяет в режиме реального времени реплицировать данные во вторичные базы данных для целей аварийного восстановления. В случае сбоя CDC позволяет быстро переключиться на вторичную базу данных, минимизируя время простоя и потерю данных.

Аудит и соответствие

CDC обеспечивает подробный контроль изменений данных, что необходимо для соблюдения нормативных требований.

Что вы построите

В этой кодовой лаборатории вы собираетесь построить конвейер данных change-data-capture (CDC) с использованием Cloud Pub/Sub, Dataproc, Python и Apache Spark. Ваш конвейер будет:

Моделируйте изменения в базе данных и публикуйте их как события в Cloud Pub/Sub — масштабируемом и надежном сервисе обмена сообщениями.
Используйте возможности Dataproc, управляемого сервиса Spark и Hadoop от Google Cloud, для обработки этих событий в режиме реального времени.

Подключив эти сервисы, вы создадите надежный конвейер, способный фиксировать и обрабатывать изменения данных по мере их возникновения, обеспечивая основу для аналитики в реальном времени, хранения данных и других критически важных приложений.

Чему вы научитесь

Как создать базовый конвейер сбора данных об изменениях
Dataproc для потоковой обработки
Cloud Pub/Sub для обмена сообщениями в реальном времени
Основы Apache Spark

Эта кодовая лаборатория сосредоточена на Dataproc и Cloud Pub/Sub. Нерелевантные концепции и блоки кода опущены и предоставлены для простого копирования и вставки.

Что вам понадобится

активная учетная запись GCP с настроенным проектом. Если у вас ее нет, вы можете зарегистрироваться на бесплатную пробную версию.
gcloud CLI установлен и настроен.
Установлен Python 3.7+ для моделирования изменений в базе данных и взаимодействия с Pub/Sub.
Базовые знания Dataproc, Cloud Pub/Sub, Apache Spark и Python.

Прежде чем начать

Выполните следующую команду в терминале, чтобы включить необходимые API:

gcloud services enable \
    dataproc.googleapis.com \
    pubsub.googleapis.com \

2. Настройте Cloud Pub/Sub

Создать тему

Эта тема будет использоваться для публикации изменений в базе данных. Задание Dataproc будет потребителем этих сообщений и будет обрабатывать сообщения для сбора данных об изменениях. Если вы хотите узнать больше о темах, вы можете прочитать официальную документацию здесь .

gcloud pubsub topics create database-changes

Создать подписку

Создайте подписку, которая будет использоваться для потребления сообщений в Pub/Sub. Чтобы узнать больше о подписках, вы можете прочитать официальную документацию здесь .

gcloud pubsub subscriptions create --topic database-changes change-subscriber

3. Моделирование изменений в базе данных

Шаги

Создайте скрипт Python (например, simulate_cdc.py ) для имитации изменений в базе данных и опубликуйте их в Pub/Sub.

from google.cloud import pubsub_v1
import json
import time
import random

project_id = "YOUR_PROJECT_ID"  # Replace with your GCP project ID
topic_id = "database-changes"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

def publish_message(data):
    data_str = json.dumps(data).encode("utf-8")
    future = publisher.publish(topic_path, data=data_str)
    print(f"Published message ID: {future.result()}")

def simulate_change():
    change_types = ["INSERT", "UPDATE", "DELETE"]
    change_type = random.choice(change_types)
    record_id = random.randint(1, 100)
    timestamp = time.time()
    change_data = {
        "change_type": change_type,
        "record_id": record_id,
        "timestamp": timestamp,
        "data": {"field1": "value1", "field2": "value2"}, #Place holder data.
    }
    publish_message(change_data)

if __name__ == "__main__":
    while True:
        simulate_change()
        time.sleep(2)  # Simulate changes every 2 seconds

Замените YOUR_PROJECT_ID на ваш фактический идентификатор проекта GCP.

Установите клиентскую библиотеку Pub/Sub:

pip install google-cloud-pubsub

Запустите скрипт на своем терминале. Этот скрипт будет работать непрерывно и публиковать сообщения каждые 2 секунды в теме Pub/Sub.

python simulate_cdc.py

После запуска скрипта, скажем, в течение 1 минуты, у вас будет достаточно сообщений в Pub/Sub для потребления. Вы можете завершить запущенный скрипт python, нажав ctrl + C или Cmd + C, в зависимости от вашей ОС.
Просмотреть опубликованные сообщения:

Откройте другой терминал и выполните следующую команду для просмотра опубликованных сообщений:

gcloud pubsub subscriptions pull --auto-ack change-subscriber

Вы должны увидеть строку таблицы, содержащую сообщение и другие поля:

{"change_type": "UPDATE", "record_id": 10, "timestamp": 1742466264.888465, "data": {"field1": "value1", "field2": "value2"}}

Объяснение

Скрипт Python имитирует изменения в базе данных, случайным образом генерируя события INSERT , UPDATE или DELETE .
Каждое изменение представлено в виде объекта JSON, содержащего тип изменения, идентификатор записи, временную метку и данные.
Скрипт использует клиентскую библиотеку Cloud Pub/Sub для публикации этих событий изменений в теме database-changes .
Команда subscriber позволяет просматривать сообщения, отправляемые в тему pub/sub.

4. Создайте учетную запись службы для Dataproc

В этом разделе вы создаете учетную запись Service, которую может использовать кластер Dataproc. Вы также назначаете необходимые разрешения, чтобы разрешить экземплярам кластера получать доступ к Cloud Pub/sub и Dataproc.

Создайте учетную запись службы:

export SERVICE_ACCOUNT_NAME="dataproc-service-account"

gcloud iam service-accounts create $SERVICE_ACCOUNT_NAME

Добавьте рабочую роль Dataproc, чтобы разрешить учетной записи службы создавать кластеры и запускать задания. Добавьте идентификатор учетной записи службы, сгенерированный в предыдущей команде, в качестве члена в следующей команде:

export PROJECT=$(gcloud info --format='value(config.project)')

gcloud projects add-iam-policy-binding $PROJECT \
        --role roles/dataproc.worker \
        --member="serviceAccount:<your-service-account-with-domain>"

Добавьте роль подписчика Pub/sub, чтобы разрешить учетной записи службы подписываться на подписку Pub/sub «change-subscriber»:

gcloud beta pubsub subscriptions add-iam-policy-binding \
        change-subscriber \
        --role roles/pubsub.subscriber \
        --member="serviceAccount:<your-service-account-with-domain"

5. Создайте кластер Dataproc

Кластер Dataproc запустит приложение spark, которое будет обрабатывать сообщения в Pub/sub. Вам понадобится учетная запись службы, созданная в предыдущем разделе. Dataproc назначает эту учетную запись службы каждому экземпляру в кластере, чтобы все экземпляры получили правильные разрешения для запуска приложения.

Для создания кластера Dataproc используйте следующую команду:

gcloud dataproc clusters create cdc-dataproc-cluster \
    --region=us-central1 \
    --zone=us-central1-a \
    --scopes=pubsub,datastore \
    --image-version=1.3 \
    --service-account="<your-service-account-with-domain-id>"

6. Отправьте задание Spark в кластер Dataproc

Приложение потоковой передачи Spark обрабатывает сообщения об изменении базы данных в Pub/sub и выводит их на консоль.

Шаги

Создайте каталог и добавьте исходный код потребителя в файл PubsubConsumer.scala.

mkdir -p dataproc-pubsub-spark-streaming/spark/src/main/scala/demo && \
touch dataproc-pubsub-spark-streaming/spark/src/main/scala/demo/PubsubConsumer.scala

package demo

import java.nio.charset.StandardCharsets
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.pubsub.{PubsubUtils, SparkGCPCredentials}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object PubsubConsumer {

  def createContext(projectID: String, checkpointDirectory: String)
    : StreamingContext = {

    // [START stream_setup]
    val sparkConf = new SparkConf().setAppName("DatabaseUpdates")
    val ssc = new StreamingContext(sparkConf, Seconds(1))

    // Set the checkpoint directory
    val yarnTags = sparkConf.get("spark.yarn.tags")
    val jobId = yarnTags.split(",").filter(_.startsWith("dataproc_job")).head
    ssc.checkpoint(checkpointDirectory + '/' + jobId)
    
    // Create stream
    val messagesStream: DStream[String] = PubsubUtils
      .createStream(
        ssc,
        projectID,
        None,
        "change-subscriber",  // Cloud Pub/Sub subscription for incoming database updates
        SparkGCPCredentials.builder.build(), StorageLevel.MEMORY_AND_DISK_SER_2)
      .map(message => new String(message.getData(), StandardCharsets.UTF_8))
    // [END stream_setup]

    processStringDStream(messagesStream)
    
        ssc
  }

  def processStringDStream(stringDStream: DStream[String]): Unit = {
    stringDStream.foreachRDD { rdd =>
      if (!rdd.isEmpty()) {
        val listOfStrings: List[String] = rdd.collect().toList
        listOfStrings.foreach(str => println(s"message received: $str"))
      } else {
        println("looking for message...")
      }
    }
  }

  def main(args: Array[String]): Unit = {
    if (args.length != 2) {
      System.err.println("arguments are not passed correctly!")
      System.exit(1)
    }

    val Seq(projectID, checkpointDirectory) = args.toSeq

    // Create Spark context
    val ssc = StreamingContext.getOrCreate(checkpointDirectory,
      () => createContext(projectID, checkpointDirectory))

    // Start streaming until we receive an explicit termination
    ssc.start()
    ssc.awaitTermination()
  }

}

Создайте и добавьте следующее в pom.xml

touch dataproc-pubsub-spark-streaming/spark/pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <jvm.options.xms>-Xms512M</jvm.options.xms>
    <jvm.options.xmx>-Xmx2048M</jvm.options.xmx>
    <jvm.options.maxpermsize>-XX:MaxPermSize=2048M</jvm.options.maxpermsize>
    <jvm.options.xx>-XX:+CMSClassUnloadingEnabled</jvm.options.xx>
  </properties>

  <groupId>dataproc-spark-demos</groupId>
  <artifactId>spark-streaming-pubsub-demo</artifactId>
  <version>1.0-SNAPSHOT</version>

  <dependencies>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.11</artifactId>
      <version>2.2.0</version>
      <scope>provided</scope>
    </dependency>

    <dependency>
      <groupId>com.google.cloud</groupId>
      <artifactId>google-cloud-datastore</artifactId>
      <version>1.34.0</version>
    </dependency>

    <dependency>
      <groupId>org.apache.bahir</groupId>
      <artifactId>spark-streaming-pubsub_2.11</artifactId>
      <version>2.2.0</version>
    </dependency>

    <dependency>
      <groupId>org.scalacheck</groupId>
      <artifactId>scalacheck_2.11</artifactId>
      <version>1.14.0</version>
      <scope>test</scope>
    </dependency>

  </dependencies>

  <build>
    <plugins>

      <plugin>
        <groupId>net.alchim31.maven</groupId>
        <artifactId>scala-maven-plugin</artifactId>
        <version>3.3.2</version>
        <executions>
          <execution>
            <id>compile</id>
            <goals>
              <goal>compile</goal>
            </goals>
            <phase>compile</phase>
          </execution>
          <execution>
            <id>test-compile</id>
            <goals>
              <goal>testCompile</goal>
            </goals>
            <phase>test-compile</phase>
          </execution>
          <execution>
            <phase>process-resources</phase>
            <goals>
              <goal>compile</goal>
            </goals>
          </execution>
        </executions>
      </plugin>

      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>3.1.1</version>
        <executions>
          <execution>
            <phase>package</phase>
            <goals>
              <goal>shade</goal>
            </goals>
            <configuration>
              <transformers>
                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                  <mainClass>demo.PubsubConsumer</mainClass>
                </transformer>
              </transformers>
              <relocations>
                <relocation>
                  <pattern>com</pattern>
                  <shadedPattern>repackaged.com</shadedPattern>
                  <includes>
                    <include>com.google.protobuf.**</include>
                    <include>com.google.common.**</include>
                  </includes>
                </relocation>
              </relocations>
            </configuration>
          </execution>
        </executions>
      </plugin>

    </plugins>
  </build>

</project>

Перейдите в каталог проекта spark и сохраните путь в переменной среды для дальнейшего использования:

cd dataproc-pubsub-spark-streaming
export REPO_ROOT=$PWD

Изменить каталог:

cd $REPO_ROOT/spark

Загрузите Java 1.8 и поместите папку в /usr/lib/jvm/. Затем измените JAVA_HOME так, чтобы он указывал на это:

export JAVA_HOME=/usr/lib/jvm/<your-java-1.8>/jre

Создайте jar-файл приложения

mvn clean package

Архив spark-streaming-pubsub-demo-1.0-SNAPSHOT.jar, содержащий код приложения и зависимости, создается в каталоге spark/target

Подать заявку на искру:

export PROJECT=$(gcloud info --format='value(config.project)')
export JAR="spark-streaming-pubsub-demo-1.0-SNAPSHOT.jar"
export SPARK_PROPERTIES="spark.dynamicAllocation.enabled=false,spark.streaming.receiver.writeAheadLog.enabled=true"
export ARGUMENTS="$PROJECT hdfs:///user/spark/checkpoint"

gcloud dataproc jobs submit spark \
    --cluster cdc-dataproc-cluster \
    --region us-central1 \
    --async \
    --jar target/$JAR \
    --max-failures-per-hour 10 \
    --properties $SPARK_PROPERTIES \
    -- $ARGUMENTS

Отобразите список активных заданий и запишите значение JOB_ID для задания:

gcloud dataproc jobs list --region=us-central1 --state-filter=active

Вывод будет выглядеть примерно так:

JOB_ID                            TYPE   STATUS
473ecb6d14e2483cb88a18988a5b2e56  spark  RUNNING

Просмотрите вывод задания, открыв следующий URL в браузере. Замените [JOB_ID] значением, указанным в предыдущем шаге.

https://console.cloud.google.com/dataproc/jobs/[JOB_ID]?region=us-central1

Вывод будет примерно следующим:

looking for message...
looking for message...
message received: {"change_type": "INSERT", "record_id": 72, "timestamp": 1745409434.969086, "data": {"field1": "value1", "field2": "value2"}}
message received: {"change_type": "INSERT", "record_id": 55, "timestamp": 1745409439.269171, "data": {"field1": "value1", "field2": "value2"}}
looking for message...
message received: {"change_type": "DELETE", "record_id": 71, "timestamp": 1745409430.673305, "data": {"field1": "value1", "field2": "value2"}}
message received: {"change_type": "UPDATE", "record_id": 15, "timestamp": 1745409432.819154, "data": {"field1": "value1", "field2": "value2"}}
message received: {"change_type": "DELETE", "record_id": 18, "timestamp": 1745409426.3570209, "data": {"field1": "value1", "field2": "value2"}}
message received: {"change_type": "INSERT", "record_id": 85, "timestamp": 1745409428.5078359, "data": {"field1": "value1", "field2": "value2"}}
message received: {"change_type": "UPDATE", "record_id": 18, "timestamp": 1745409441.436026, "data": {"field1": "value1", "field2": "value2"}}
looking for message...

Задание потоковой передачи Spark, запущенное в Dataproc, извлекает сообщения из Pub/sub, обрабатывает их и выводит вывод на консоль.

Завершение задания: Выполните следующую команду, чтобы завершить задание. Замените JOB_ID на тот же, который мы отметили ранее

gcloud dataproc jobs kill [JOB_ID] --region us-central1 --quiet

Поздравляем! Вы только что создали мощный конвейер CDC, который фиксирует изменения базы данных в Pub/sub и обрабатывает их с помощью потоковой передачи Spark, запущенной в Cloud Dataproc.

7. Уборка

Очистите все созданные вами ресурсы, чтобы вам не пришлось платить за них в будущем. Самый простой способ избавиться от выставления счетов — удалить проект, созданный вами для руководства. В качестве альтернативы вы можете удалить отдельные ресурсы.

Чтобы удалить отдельные ресурсы, выполните следующие команды:

gcloud dataproc clusters delete cdc-dataproc-cluster --quiet
gcloud pubsub topics delete database-changes --quiet
gcloud pubsub subscriptions delete change-subscriber --quiet
gcloud iam service-accounts delete <your-service-account-with-domain> --quiet

8. Поздравления

Поздравляем, вы только что завершили практическую кодовую лабораторию, которая демонстрирует, как построить надежный конвейер данных в реальном времени с использованием Google Cloud Platform. Давайте подведем итоги того, чего вы достигли:

Моделирование сбора измененных данных (CDC): вы изучили основы CDC и реализовали скрипт Python для моделирования изменений в базе данных, генерируя события, представляющие изменения данных в реальном времени.
Leveraged Cloud Pub/Sub: Вы настраиваете темы и подписки Cloud Pub/Sub, предоставляя масштабируемую и надежную службу обмена сообщениями для потоковой передачи ваших событий CDC. Вы публикуете свои смоделированные изменения базы данных в Pub/Sub, создавая поток данных в реальном времени.
Обработанные данные с Dataproc и Spark: вы подготовили кластер Dataproc и развернули задание Spark Streaming для потребления сообщений из вашей подписки Pub/Sub. Вы обработали и преобразовали входящие события CDC в режиме реального времени, отобразив результаты в журналах заданий Dataproc.
Построили сквозной конвейер в реальном времени: вы успешно интегрировали эти сервисы для создания полного конвейера данных, который захватывает, транслирует и обрабатывает изменения данных в реальном времени. Вы приобрели практический опыт в построении системы, которая может обрабатывать непрерывные потоки данных.
Использован коннектор Spark Pub/Sub: вы успешно настроили кластер Dataproc для использования коннектора Spark Pub/Sub, который имеет решающее значение для Spark Structured Streaming для чтения данных из Pub/Sub.

Теперь у вас есть прочная основа для создания более сложных и сложных конвейеров данных для различных приложений, включая аналитику в реальном времени, хранилища данных и архитектуры микросервисов. Продолжайте исследовать и строить!

Создание сбора данных об изменениях с использованием Dataproc и Cloud Pub/Sub

1. Введение

Что такое сбор измененных данных?

Миграция данных

Хранилище данных и аналитика в реальном времени

Аварийное восстановление и высокая доступность

Аудит и соответствие

Что вы построите

Чему вы научитесь

Что вам понадобится

Прежде чем начать

2. Настройте Cloud Pub/Sub

Создать тему

Создать подписку

3. Моделирование изменений в базе данных

Шаги

Объяснение

4. Создайте учетную запись службы для Dataproc

5. Создайте кластер Dataproc

6. Отправьте задание Spark в кластер Dataproc

Шаги

7. Уборка

8. Поздравления

Справочные документы