En esta página, se muestra cómo crear canalizaciones de Dataflow que consumen y reenvían datos de cambios de Spanner mediante el uso de transmisiones de cambios. Puedes usar el código de ejemplo en esta página para crear canalizaciones personalizadas.
Conceptos básicos
Los siguientes son algunos conceptos básicos de las canalizaciones de Dataflow para flujos de cambios.
Dataflow
Dataflow es un servicio sin servidores, rápido y rentable que admite procesamiento por lotes y de transmisión. Proporciona portabilidad con trabajos de procesamiento escritos mediante las bibliotecas de código abierto de Apache Beam y automatiza el aprovisionamiento de infraestructura y la administración de clústeres. Dataflow proporciona una transmisión casi en tiempo real cuando se leen datos de flujos de cambios.
Puedes usar Dataflow para consumir flujos de cambios de Spanner con el conector de Spanner, que ofrece una abstracción en la API de Spanner para consultar flujos de cambios. Con este conector, no tienes que administrar el ciclo de vida de la partición de los flujos de cambios, lo cual es necesario cuando usas la API de Spanner directamente. El conector te proporciona una transmisión de registros de cambios de datos para que puedas enfocarte más en la lógica de la aplicación y menos en los detalles específicos de la API y la partición dinámica del flujo de cambios. Recomendamos usar el conector SpannerIO en lugar de la API de Spanner en la mayoría de las circunstancias en las que necesites leer datos de flujo de cambios.
Las plantillas de Dataflow son canalizaciones precompiladas que implementan casos de uso comunes. Consulta Plantillas de Dataflow para obtener una descripción general.
Canalización de Dataflow
Una canalización de Dataflow de flujos de cambios de Spanner consta de cuatro partes principales:
- Una base de datos de Spanner con un flujo de cambios
- El conector SpannerIO
- Transformaciones y receptores definidos por el usuario
- Escritor de E/S de un receptor
Cada uno de ellos se analiza con más detalle a continuación.
Flujo de cambios de Spanner
Para obtener detalles sobre cómo crear un flujo de cambios, consulta Crea un flujo de cambios.
Conector de SpannerIO de Apache Beam
Este es el conector SpannerIO descrito antes. Es un conector de E/S de origen que emite un PCollection de registros de cambios de datos a etapas posteriores de la canalización. El tiempo del evento para cada registro de cambio de datos emitido será la marca de tiempo de confirmación. Ten en cuenta que los registros emitidos están desordenados y que el conector de SpannerIO garantiza que no habrá registros tardíos.
Cuando se trabaja con flujos de cambios, Dataflow usa puntos de control. Como resultado, cada trabajador puede esperar hasta el intervalo de punto de control configurado para almacenar en búfer los cambios antes de enviarlos para su procesamiento posterior.
Transformaciones definidas por el usuario
Una transformación definida por el usuario le permite agregar, transformar o modificar datos de procesamiento dentro de una canalización de Dataflow. Los casos de uso comunes para esto son quitar la información de identificación personal, satisfacer los requisitos de formato de datos descendentes y el ordenamiento. Consulta la documentación oficial de Apache Beam para obtener la guía de programación sobre las transforms.
Escritor de E/S del receptor de Apache Beam
Apache Beam contiene transformaciones de E/S integradas que se pueden usar para escribir desde una canalización de Dataflow en un receptor de datos como BigQuery. Los receptores de datos más comunes tienen compatibilidad nativa.
Plantillas de Dataflow
Las plantillas de Dataflow proporcionan una manera fácil de crear trabajos de Dataflow basados en imágenes de Docker compiladas con anterioridad para casos de uso comunes a través de la consola de Google Cloud, Google Cloud CLI o llamadas a la API de REST.
Para las flujos de cambios de Spanner, proporcionamos tres plantillas de Flex de Dataflow:
Crea una canalización de Dataflow
En esta sección, se aborda la configuración inicial del conector y se proporcionan muestras de integraciones comunes con la función de flujos de cambios de Spanner.
Para seguir estos pasos, necesitas un entorno de desarrollo de Java para Dataflow. Para obtener más información, consulta Crea una canalización de Dataflow con Java.
Crear transmisión de cambios
Si deseas obtener detalles para crear un flujo de cambios, consulta Cómo crear un flujo de cambios. Para continuar con los siguientes pasos, debes tener una base de datos de Spanner con un flujo de cambios configurado.
Otorga privilegios de control de acceso detallados
Si esperas que algún usuario de control de acceso detallado ejecute el trabajo de Dataflow, asegúrate de que se les otorgue acceso a una función de base de datos que tenga el privilegio SELECT en el flujo de cambios y el privilegio EXECUTE en la función de valor de tabla del flujo de cambios. Además, asegúrate de que la principal especifique la función de la base de datos en la configuración de SpannerIO o en la plantilla de Flex de Dataflow.
Para obtener más información, consulta Acerca del control de acceso detallado.
Agrega el conector SpannerIO como dependencia
El conector SpannerIO de Apache Beam encapsula la complejidad de consumir los flujos de cambios directamente a través de la API de Cloud Spanner, lo que emite una PCollection de registros de datos de flujos de cambios a etapas posteriores de la canalización.
Estos objetos se pueden consumir en otras etapas de la canalización de Dataflow del usuario. La integración del flujo de cambios forma parte del conector SpannerIO. Para poder usar el conector SpannerIO, debes agregar la dependencia al archivo pom.xml:
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
<version>${beam-version}</version> <!-- available from version 2.38.0 -->
</dependency>
Crea una base de datos de metadatos
El conector debe realizar un seguimiento de cada partición cuando ejecuta la canalización de Apache Beam. Mantiene estos metadatos en una tabla de Spanner creada por el conector durante la inicialización. Especifica la base de datos en la que se creará esta tabla cuando configures el conector.
Como se describe en Prácticas recomendadas para los flujos de cambios, recomendamos crear una base de datos nueva para este propósito, en lugar de permitir que el conector use la base de datos de tu aplicación para almacenar su tabla de metadatos.
El propietario de un trabajo de Dataflow que usa el conector SpannerIO debe tener los siguientes permisos de IAM configurados con esta base de datos de metadatos:
spanner.databases.updateDdlspanner.databases.beginReadOnlyTransactionspanner.databases.beginOrRollbackReadWriteTransactionspanner.databases.readspanner.databases.selectspanner.databases.writespanner.sessions.createspanner.sessions.get
Configura el conector
El conector de flujos de cambios de Spanner se puede configurar de la siguiente manera:
SpannerConfig spannerConfig = SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role"); // Needed for fine-grained access control only
Timestamp startTime = Timestamp.now();
Timestamp endTime = Timestamp.ofTimeSecondsAndNanos(
startTime.getSeconds() + (10 * 60),
startTime.getNanos()
);
SpannerIO
.readChangeStream()
.withSpannerConfig(spannerConfig)
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-meta-instance-id")
.withMetadataDatabase("my-meta-database-id")
.withMetadataTable("my-meta-table-name")
.withRpcPriority(RpcPriority.MEDIUM)
.withInclusiveStartAt(startTime)
.withInclusiveEndAt(endTime);
Las siguientes son descripciones de las opciones de readChangeStream():
Configuración de Spanner (obligatoria)
Se usa para configurar el proyecto, la instancia y la base de datos donde se creó el flujo de cambios y desde donde se debe realizar la consulta. De manera opcional, también especifica el rol de base de datos que se usará cuando la principal de IAM que ejecuta el trabajo de Dataflow sea un usuario de control de acceso detallado. El trabajo asume este rol de la base de datos para acceder al flujo de cambios. Para obtener más información, consulta Acerca del control de acceso detallado.
Nombre del flujo de cambios (obligatorio)
Este nombre identifica de forma única el flujo de cambios. El nombre debe ser el mismo que se usó al crearlo.
ID de instancia de metadatos (opcional)
Esta es la instancia en la que se almacenarán los metadatos que usa el conector para controlar el consumo de los datos de la API de flujos de cambios.
ID de la base de datos de metadatos (obligatorio)
Esta es la base de datos en la que se almacenarán los metadatos que usa el conector para controlar el consumo de los datos de la API de flujos de cambios.
Nombre de la tabla de metadatos (opcional)
Solo debe usarse cuando se actualiza una canalización existente.
Este es el nombre preexistente de la tabla de metadatos que usará el conector. El conector lo usa para almacenar los metadatos y controlar el consumo de los datos de la API de flujo de cambios. Si se omite esta opción, Spanner crea una tabla nueva con un nombre generado cuando se inicializa el conector.
Prioridad de RPC (opcional)
La prioridad de la solicitud que se usará para las consultas del flujo de cambios. Si se omite este parámetro, se usará high
priority.
InclusiveStartAt (obligatorio)
Los cambios de la marca de tiempo dada se muestran al llamador.
InclusiveEndAt (opcional)
Los cambios hasta la marca de tiempo dada se muestran al llamador. Si se omite este parámetro, los cambios se emitirán de forma indefinida.
Agrega transformaciones y receptores para procesar datos modificados
Con los pasos anteriores completados, el conector SpannerIO configurado está listo para emitir una PCollection de objetos DataChangeRecord.
Consulta Ejemplo de transformaciones y receptores para ver varias configuraciones de canalización de muestra que procesan estos datos transmitidos de varias maneras.
Ten en cuenta que los registros de flujos de cambios emitidos por el conector de SpannerIO no están ordenados. Esto se debe a que las PCollections no ofrecen ninguna garantía de ordenamiento. Si necesitas una transmisión ordenada, debes agrupar y ordenar los registros como transformaciones en tus canalizaciones: consulta Muestra: Ordenar por clave. Puedes extender este ejemplo para ordenar los registros según cualquier campo de registros, por ejemplo, por IDs de transacción.
Transformaciones y receptores de ejemplo
Puedes definir tus propias transformaciones y especificar receptores en los que escribir los datos. La documentación de Apache Beam ofrece una gran cantidad de transforms que se pueden aplicar, así como conectores de E/S listos para usar con el objetivo de escribir los datos en sistemas externos.
Muestra: Ordena por clave
Esta muestra de código emite registros de cambios de datos ordenados por marca de tiempo de confirmación y agrupados por claves primarias mediante el conector de Dataflow.
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(ParDo.of(new BreakRecordByModFn()))
.apply(ParDo.of(new KeyByIdFn()))
.apply(ParDo.of(new BufferKeyUntilOutputTimestamp()))
// Subsequent processing goes here
En esta muestra de código, se usan estados y cronómetros para almacenar en búfer los registros de cada clave, y se establece el tiempo de vencimiento del temporizador en un tiempo T configurado por el usuario en el futuro (definido en la función BufferKeyUntilOutputTimestamp). Cuando la marca de agua de Dataflow pasa el tiempo T, este código limpia todos los registros del búfer con una marca de tiempo inferior a T, ordena estos registros por marca de tiempo de confirmación y da como resultado un par clave-valor en el que sucede lo siguiente:
- La clave es la clave de entrada, que es la clave primaria con hash para un array de bucket de tamaño 1,000.
- El valor son los registros de cambios de datos ordenados que se almacenaron en el búfer para la clave.
Para cada clave, tenemos las siguientes garantías:
- Se garantiza que los cronómetros se activan en el orden de la marca de tiempo de vencimiento.
- Las etapas descendentes garantizan la recepción de los elementos en el mismo orden en que se produjeron.
Por ejemplo, supongamos que para una clave con el valor 100, el temporizador se activa en T1 y T10 respectivamente, lo que produce un paquete de registros de cambios de datos en cada marca de tiempo. Debido a que los registros de cambios en los datos generados en T1 se produjeron antes que los registros de cambios en los datos generados en T10, también se garantiza que los registros de cambios en los datos enviados en T1 se recibirán en la siguiente etapa antes de que los registros de cambios en los datos enviados en T10. Este mecanismo nos ayuda a garantizar un orden estricto de la marca de tiempo de confirmación por clave primaria para el procesamiento posterior.
Este proceso se repetirá hasta que finalice la canalización y se hayan procesado todos los registros de cambios de datos (o se repetirá indefinidamente si no se especifica una hora de finalización).
Ten en cuenta que esta muestra de código usa estados y temporizadores, en lugar de ventanas, para ordenar por clave. La lógica es que no se garantiza que las ventanas se procesen en orden. Esto significa que las ventanas más antiguas se pueden procesar más tarde que las más recientes, lo que podría provocar un procesamiento desordenado.
BreakRecordByModFn
Cada registro de cambios en los datos puede contener varias modificaciones. Cada modificación representa una inserción, actualización o eliminación de un solo valor de clave primaria. Esta función divide cada registro de cambios de datos en registros separados de cambios de datos, uno por modificación.
private static class BreakRecordByModFn extends DoFn<DataChangeRecord,
DataChangeRecord> {
@ProcessElement
public void processElement(
@Element DataChangeRecord record, OutputReceiver<DataChangeRecord>
outputReceiver) {
record.getMods().stream()
.map(
mod ->
new DataChangeRecord(
record.getPartitionToken(),
record.getCommitTimestamp(),
record.getServerTransactionId(),
record.isLastRecordInTransactionInPartition(),
record.getRecordSequence(),
record.getTableName(),
record.getRowType(),
Collections.singletonList(mod),
record.getModType(),
record.getValueCaptureType(),
record.getNumberOfRecordsInTransaction(),
record.getNumberOfPartitionsInTransaction(),
record.getTransactionTag(),
record.isSystemTransaction(),
record.getMetadata()))
.forEach(outputReceiver::output);
}
}
KeyByIdFn
Esta función toma un DataChangeRecord y da como resultado un DataChangeRecord con clave primaria de Spanner codificada en un valor de número entero.
private static class KeyByIdFn extends DoFn<DataChangeRecord, KV<String, DataChangeRecord>> {
// NUMBER_OF_BUCKETS should be configured by the user to match their key cardinality
// Here, we are choosing to hash the Spanner primary keys to a bucket index, in order to have a deterministic number
// of states and timers for performance purposes.
// Note that having too many buckets might have undesirable effects if it results in a low number of records per bucket
// On the other hand, having too few buckets might also be problematic, since many keys will be contained within them.
private static final int NUMBER_OF_BUCKETS = 1000;
@ProcessElement
public void processElement(
@Element DataChangeRecord record,
OutputReceiver<KV<String, DataChangeRecord>> outputReceiver) {
int hashCode = (int) record.getMods().get(0).getKeysJson().hashCode();
// Hash the received keys into a bucket in order to have a
// deterministic number of buffers and timers.
String bucketIndex = String.valueOf(hashCode % NUMBER_OF_BUCKETS);
outputReceiver.output(KV.of(bucketIndex, record));
}
}
BufferKeyUntilOutputTimestamp
Los temporizadores y los búferes son por clave. Esta función almacena en búfer cada registro de cambios de datos hasta que la marca de agua pasa la marca de tiempo en la que queremos generar los registros de cambios de datos almacenados en búfer.
En este código, se usa un temporizador que se repite indefinidamente para determinar cuándo se debe vaciar el búfer:
- Cuando ve un registro de cambios de datos de una clave por primera vez, configura el temporizador para que se active en la marca de tiempo de confirmación del registro de cambio de datos +
incrementIntervalSeconds(una opción configurable por el usuario). - Cuando el temporizador se activa, agrega todos los registros de cambios de datos en el búfer con una marca de tiempo menor que el tiempo de vencimiento del temporizador en
recordsToOutput. Si el búfer tiene registros de cambios de datos cuya marca de tiempo es mayor o igual que la hora de vencimiento del temporizador, agrega esos registros de cambios de datos al búfer en lugar de generarlos. Luego, establece el próximo temporizador en el tiempo de vencimiento actual y enincrementIntervalInSeconds. - Si
recordsToOutputno está vacío, la función ordena los registros de cambios de datos enrecordsToOutputpor marca de tiempo de confirmación y por ID de transacción, y, luego, los muestra.
private static class BufferKeyUntilOutputTimestamp extends
DoFn<KV<String, DataChangeRecord>, KV<String, Iterable<DataChangeRecord>>> {
private static final Logger LOG =
LoggerFactory.getLogger(BufferKeyUntilOutputTimestamp.class);
private final long incrementIntervalInSeconds = 2;
private BufferKeyUntilOutputTimestamp(long incrementIntervalInSeconds) {
this.incrementIntervalInSeconds = incrementIntervalInSeconds;
}
@SuppressWarnings("unused")
@TimerId("timer")
private final TimerSpec timerSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
@StateId("buffer")
private final StateSpec<BagState<DataChangeRecord>> buffer = StateSpecs.bag();
@StateId("keyString")
private final StateSpec<ValueState<String>> keyString =
StateSpecs.value(StringUtf8Coder.of());
@ProcessElement
public void process(
@Element KV<String, DataChangeRecord> element,
@StateId("buffer") BagState<DataChangeRecord> buffer,
@TimerId("timer") Timer timer,
@StateId("keyString") ValueState<String> keyString) {
buffer.add(element.getValue());
// Only set the timer if this is the first time we are receiving a data change
// record with this key.
String elementKey = keyString.read();
if (elementKey == null) {
Instant commitTimestamp =
new Instant(element.getValue().getCommitTimestamp().toSqlTimestamp());
Instant outputTimestamp =
commitTimestamp.plus(Duration.standardSeconds(incrementIntervalInSeconds));
timer.set(outputTimestamp);
keyString.write(element.getKey());
}
}
@OnTimer("timer")
public void onExpiry(
OnTimerContext context,
@StateId("buffer") BagState<DataChangeRecord> buffer,
@TimerId("timer") Timer timer,
@StateId("keyString") ValueState<String> keyString) {
if (!buffer.isEmpty().read()) {
String elementKey = keyString.read();
final List<DataChangeRecord> records =
StreamSupport.stream(buffer.read().spliterator(), false)
.collect(Collectors.toList());
buffer.clear();
List<DataChangeRecord> recordsToOutput = new ArrayList<>();
for (DataChangeRecord record : records) {
Instant recordCommitTimestamp =
new Instant(record.getCommitTimestamp().toSqlTimestamp());
final String recordString =
record.getMods().get(0).getNewValuesJson().isEmpty()
? "Deleted record"
: record.getMods().get(0).getNewValuesJson();
// When the watermark passes time T, this means that all records with
// event time < T have been processed and successfully committed. Since the
// timer fires when the watermark passes the expiration time, we should
// only output records with event time < expiration time.
if (recordCommitTimestamp.isBefore(context.timestamp())) {
LOG.info(
"Outputting record with key {} and value {} at expiration " +
"timestamp {}",
elementKey,
recordString,
context.timestamp().toString());
recordsToOutput.add(record);
} else {
LOG.info(
"Expired at {} but adding record with key {} and value {} back to " +
"buffer due to commit timestamp {}",
context.timestamp().toString(),
elementKey,
recordString,
recordCommitTimestamp.toString());
buffer.add(record);
}
}
// Output records, if there are any to output.
if (!recordsToOutput.isEmpty()) {
// Order the records in place, and output them. The user would need
// to implement DataChangeRecordComparator class that sorts the
// data change records by commit timestamp and transaction ID.
Collections.sort(recordsToOutput, new DataChangeRecordComparator());
context.outputWithTimestamp(
KV.of(elementKey, recordsToOutput), context.timestamp());
LOG.info(
"Expired at {}, outputting records for key {}",
context.timestamp().toString(),
elementKey);
} else {
LOG.info("Expired at {} with no records", context.timestamp().toString());
}
}
Instant nextTimer = context.timestamp().plus(Duration.standardSeconds(incrementIntervalInSeconds));
if (buffer.isEmpty() != null && !buffer.isEmpty().read()) {
LOG.info("Setting next timer to {}", nextTimer.toString());
timer.set(nextTimer);
} else {
LOG.info(
"Timer not being set since the buffer is empty: ");
keyString.clear();
}
}
}
Cómo pedir transacciones
Esta canalización se puede cambiar a orden por ID de transacción y marca de tiempo de confirmación. Para ello, almacena en búfer los registros de cada ID de transacción o par de marca de tiempo de confirmación, en lugar de cada clave de Spanner. Esta acción requiere modificar el código de KeyByIdFn.
Muestra: Ensambla transacciones
Esta muestra de código lee los registros de cambios de datos, reúne todos los registros de cambios de datos que pertenecen a la misma transacción en un solo elemento y genera un resultado en él. Ten en cuenta que las transacciones resultantes de este código de muestra no están ordenadas por marca de tiempo de confirmación.
Esta muestra de código usa búferes para ensamblar transacciones de registros de cambios de datos. Cuando recibe un registro de cambios de datos que pertenece a una transacción por primera vez, lee el campo numberOfRecordsInTransaction en el registro de cambios de datos, que describe la cantidad esperada de registros de cambios de datos que pertenecen a esa transacción. Almacena en búfer los registros de cambios de datos que pertenecen a esa transacción hasta que la cantidad de registros almacenados en el búfer coincida con numberOfRecordsInTransaction, tras lo cual se mostrarán los registros de cambios de datos agrupados.
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(ParDo.of(new KeyByTransactionIdFn()))
.apply(ParDo.of(new TransactionBoundaryFn()))
// Subsequent processing goes here
KeyByTransactionIdFn
Esta función toma un DataChangeRecord y da como resultado un DataChangeRecord con clave por el ID de transacción.
private static class KeyByTransactionIdFn extends DoFn<DataChangeRecord, KV<String, DataChangeRecord>> {
@ProcessElement
public void processElement(
@Element DataChangeRecord record,
OutputReceiver<KV<String, DataChangeRecord>> outputReceiver) {
outputReceiver.output(KV.of(record.getServerTransactionId(), record));
}
}
TransactionBoundaryFn
Los búferes TransactionBoundaryFn recibieron pares clave-valor de {TransactionId, DataChangeRecord} de KeyByTransactionIdFn y los almacena en búfer en grupos basados en TransactionId. Cuando la cantidad de registros almacenados en búfer es igual a la cantidad de registros contenidos en la transacción completa, esta función ordena los objetos DataChangeRecord del grupo por secuencia de registros y genera un par clave-valor de {CommitTimestamp, TransactionId}, Iterable<DataChangeRecord>.
Aquí, suponemos que SortKey es una clase definida por el usuario que representa un par de {CommitTimestamp, TransactionId}. Consulta la implementación de muestra para SortKey.
private static class TransactionBoundaryFn extends DoFn<KV<String, DataChangeRecord>, KV<SortKey, Iterable<DataChangeRecord>>> {
@StateId("buffer")
private final StateSpec<BagState<DataChangeRecord>> buffer = StateSpecs.bag();
@StateId("count")
private final StateSpec<ValueState<Integer>> countState = StateSpecs.value();
@ProcessElement
public void process(
ProcessContext context,
@StateId("buffer") BagState<DataChangeRecord> buffer,
@StateId("count") ValueState<Integer> countState) {
final KV<String, DataChangeRecord> element = context.element();
final DataChangeRecord record = element.getValue();
buffer.add(record);
int count = (countState.read() != null ? countState.read() : 0);
count = count + 1;
countState.write(count);
if (count == record.getNumberOfRecordsInTransaction()) {
final List<DataChangeRecord> sortedRecords =
StreamSupport.stream(buffer.read().spliterator(), false)
.sorted(Comparator.comparing(DataChangeRecord::getRecordSequence))
.collect(Collectors.toList());
final Instant commitInstant =
new Instant(sortedRecords.get(0).getCommitTimestamp().toSqlTimestamp()
.getTime());
context.outputWithTimestamp(
KV.of(
new SortKey(sortedRecords.get(0).getCommitTimestamp(),
sortedRecords.get(0).getServerTransactionId()),
sortedRecords),
commitInstant);
buffer.clear();
countState.clear();
}
}
}
Muestra: Filtra por etiqueta de transacción
Cuando se etiqueta una transacción que modifica datos del usuario, la etiqueta correspondiente y su tipo se almacenan como parte de DataChangeRecord. Estos ejemplos muestran cómo filtrar registros de flujos de cambios según las etiquetas de transacción definidas por el usuario, así como las etiquetas del sistema:
Filtrado de etiquetas definidas por el usuario para my-tx-tag:
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(Filter.by(record ->
!record.isSystemTransaction()
&& record.getTransactionTag().equalsIgnoreCase("my-tx-tag")))
// Subsequent processing goes here
Filtrado de etiquetas del sistema/auditoría de TTL:
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(Filter.by(record ->
record.isSystemTransaction()
&& record.getTransactionTag().equals("RowDeletionPolicy")))
// Subsequent processing goes here
Muestra: Recuperar fila completa
Este ejemplo funciona con una tabla de Spanner llamada Singer que tiene la siguiente definición:
CREATE TABLE Singers (
SingerId INT64 NOT NULL,
FirstName STRING(1024),
LastName STRING(1024)
) PRIMARY KEY (SingerId);
En el modo de captura de valores OLD_AND_NEW_VALUES predeterminado de los flujos de cambios, cuando hay una actualización en una fila de Spanner, el registro de cambios de datos recibido contendrá solo las columnas que se modificaron. Las columnas con seguimiento, pero sin cambios no se incluirán en el registro. La clave primaria de la modificación se puede usar para leer una instantánea de Spanner en la marca de tiempo de confirmación del registro de cambios de datos para recuperar las columnas sin cambios o incluso la fila completa.
Ten en cuenta que es posible que debas cambiar la política de retención de la base de datos a un valor mayor o igual que la política de retención del flujo de cambios para que la lectura de la instantánea se realice correctamente.
Además, ten en cuenta que usar el tipo de captura del valor NEW_ROW es la forma recomendada y más eficiente de hacerlo, ya que muestra todas las columnas con seguimiento de la fila de forma predeterminada y no requiere una instantánea adicional de lectura en Spanner.
SpannerConfig spannerConfig = SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role"); // Needed for fine-grained access control only
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(spannerConfig)
// Assume we have a change stream "my-change-stream" that watches Singers table.
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(ParDo.of(new ToFullRowJsonFn(spannerConfig)))
// Subsequent processing goes here
ToFullRowJsonFn
Esta transformación realizará una lectura inactiva en la marca de tiempo de confirmación de cada registro recibido y asignará la fila completa a JSON.
public class ToFullRowJsonFn extends DoFn<DataChangeRecord, String> {
// Since each instance of this DoFn will create its own session pool and will
// perform calls to Spanner sequentially, we keep the number of sessions in
// the pool small. This way, we avoid wasting resources.
private static final int MIN_SESSIONS = 1;
private static final int MAX_SESSIONS = 5;
private final String projectId;
private final String instanceId;
private final String databaseId;
private transient DatabaseClient client;
private transient Spanner spanner;
public ToFullRowJsonFn(SpannerConfig spannerConfig) {
this.projectId = spannerConfig.getProjectId().get();
this.instanceId = spannerConfig.getInstanceId().get();
this.databaseId = spannerConfig.getDatabaseId().get();
}
@Setup
public void setup() {
SessionPoolOptions sessionPoolOptions = SessionPoolOptions
.newBuilder()
.setMinSessions(MIN_SESSIONS)
.setMaxSessions(MAX_SESSIONS)
.build();
SpannerOptions options = SpannerOptions
.newBuilder()
.setProjectId(projectId)
.setSessionPoolOption(sessionPoolOptions)
.build();
DatabaseId id = DatabaseId.of(projectId, instanceId, databaseId);
spanner = options.getService();
client = spanner.getDatabaseClient(id);
}
@Teardown
public void teardown() {
spanner.close();
}
@ProcessElement
public void process(
@Element DataChangeRecord element,
OutputReceiver<String> output) {
com.google.cloud.Timestamp commitTimestamp = element.getCommitTimestamp();
element.getMods().forEach(mod -> {
JSONObject keysJson = new JSONObject(mod.getKeysJson());
JSONObject newValuesJson = new JSONObject(mod.getNewValuesJson());
ModType modType = element.getModType();
JSONObject jsonRow = new JSONObject();
long singerId = keysJson.getLong("SingerId");
jsonRow.put("SingerId", singerId);
if (modType == ModType.INSERT) {
// For INSERT mod, get non-primary key columns from mod.
jsonRow.put("FirstName", newValuesJson.get("FirstName"));
jsonRow.put("LastName", newValuesJson.get("LastName"));
} else if (modType == ModType.UPDATE) {
// For UPDATE mod, get non-primary key columns by doing a snapshot read using the primary key column from mod.
try (ResultSet resultSet = client
.singleUse(TimestampBound.ofReadTimestamp(commitTimestamp))
.read(
"Singers",
KeySet.singleKey(com.google.cloud.spanner.Key.of(singerId)),
Arrays.asList("FirstName", "LastName"))) {
if (resultSet.next()) {
jsonRow.put("FirstName", resultSet.isNull("FirstName") ?
JSONObject.NULL : resultSet.getString("FirstName"));
jsonRow.put("LastName", resultSet.isNull("LastName") ?
JSONObject.NULL : resultSet.getString("LastName"));
}
}
} else {
// For DELETE mod, there is nothing to do, as we already set SingerId.
}
output.output(jsonRow.toString());
});
}
}
Este código crea un cliente de base de datos de Spanner para realizar la recuperación de la fila completa, configura el grupo de sesiones para que tenga solo unas pocas sesiones y realiza lecturas en una instancia de ToFullRowJsonFn de forma secuencial. Dataflow se asegura de generar muchas instancias de esta función, cada una con su propio grupo de clientes.
Muestra: Spanner a Pub/Sub
En esta situación, el emisor transmite los registros a Pub/Sub lo más rápido posible, sin ninguna agrupación ni agregación. Esta es una buena opción para activar el procesamiento posterior, como transmitir todas las filas nuevas insertadas en una tabla de Spanner a Pub/Sub para su procesamiento posterior.
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(MapElements.into(TypeDescriptors.strings()).via(Object::toString))
.apply(PubsubIO.writeStrings().to("my-topic"));
Ten en cuenta que el receptor de Pub/Sub se puede configurar para garantizar la semántica exactamente una vez.
Muestra: Spanner a Cloud Storage
En esta situación, el llamador agrupa todos los registros dentro de una ventana determinada y guarda el grupo en archivos separados de Cloud Storage. Esta es una buena opción para las estadísticas y el archivado de un momento determinado, que es independiente del período de retención de Spanner.
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(MapElements.into(TypeDescriptors.strings()).via(Object::toString))
.apply(Window.into(FixedWindows.of(Duration.standardMinutes(1))))
.apply(TextIO
.write()
.to("gs://my-bucket/change-stream-results-")
.withSuffix(".txt")
.withWindowedWrites()
.withNumShards(1));
Ten en cuenta que el receptor de Cloud Storage proporciona semántica de al menos una vez de forma predeterminada. Con procesamiento adicional, se puede modificar para que tenga una semántica del tipo “exactamente una vez”.
También proporcionamos una plantilla de Dataflow para este caso de uso: consulta Conecta flujos de cambios a Cloud Storage.
Muestra: Spanner a BigQuery (tabla de registro)
Aquí, las transmisiones del emisor cambian los registros a BigQuery. Cada registro de cambios en los datos se refleja como una fila en BigQuery. Esta es una buena opción para las estadísticas. Este código usa las funciones que se definieron anteriormente en la sección Recuperar fila completa para recuperar la fila completa del registro y escribirla en BigQuery.
SpannerConfig spannerConfig = SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role"); // Needed for fine-grained access control only
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(spannerConfig)
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(ParDo.of(new ToFullRowJsonFn(spannerConfig)))
.apply(BigQueryIO
.<String>write()
.to("my-bigquery-table")
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(Write.WriteDisposition.WRITE_APPEND)
.withSchema(new TableSchema().setFields(Arrays.asList(
new TableFieldSchema()
.setName("SingerId")
.setType("INT64")
.setMode("REQUIRED"),
new TableFieldSchema()
.setName("FirstName")
.setType("STRING")
.setMode("REQUIRED"),
new TableFieldSchema()
.setName("LastName")
.setType("STRING")
.setMode("REQUIRED")
)))
.withAutoSharding()
.optimizedWrites()
.withFormatFunction((String element) -> {
ObjectMapper objectMapper = new ObjectMapper();
JsonNode jsonNode = null;
try {
jsonNode = objectMapper.readTree(element);
} catch (IOException e) {
e.printStackTrace();
}
return new TableRow()
.set("SingerId", jsonNode.get("SingerId").asInt())
.set("FirstName", jsonNode.get("FirstName").asText())
.set("LastName", jsonNode.get("LastName").asText());
}
)
);
Ten en cuenta que el receptor de BigQuery proporciona semántica de al menos una vez de forma predeterminada. Con procesamiento adicional, se puede modificar para que tenga una semántica del tipo “exactamente una vez”.
También proporcionamos una plantilla de Dataflow para este caso de uso. Consulta Conecta flujos de cambios a BigQuery.
Supervisar una canalización
Hay dos clases de métricas disponibles para supervisar una canalización de flujo de cambios de Dataflow.
Métricas estándar de Dataflow
Dataflow proporciona varias métricas para asegurarse de que tu trabajo esté en buen estado, como actualización de datos, retraso del sistema, capacidad de procesamiento del trabajo, uso de CPU del trabajador y mucho más. Puedes encontrar más información en Usa Monitoring para las canalizaciones de Dataflow.
Para las canalizaciones de flujos de cambios, hay dos métricas principales que se deben tener en cuenta: la latencia del sistema y la actualización de los datos.
La latencia del sistema te indicará la duración máxima actual (en segundos) durante la cual un elemento de datos se está procesando o en espera de procesamiento.
La actualización de los datos te mostrará el tiempo transcurrido entre este momento (en tiempo real) y la marca de agua de salida. La marca de agua de salida con el tiempo T indica que todos los elementos con una hora de evento (estrictamente) anterior a T se procesaron para el procesamiento. En otras palabras, la actualidad de los datos mide qué tan actualizada está la canalización en relación con el procesamiento de los eventos que recibió.
Si la canalización no cuenta con los recursos suficientes, puedes ver ese efecto en estas dos métricas. La latencia del sistema aumentará, ya que los elementos deben esperar más tiempo para poder procesarse. La actualidad de los datos también aumentará, ya que la canalización no podrá mantener el ritmo de la cantidad de datos recibidos.
Métricas personalizadas de los flujos de cambios
Estas métricas se exponen en Cloud Monitoring y se incluyen las siguientes:
- Latencia en buckets (histograma) entre un registro que se confirma en Spanner y que el conector emite en una PCollection. Esta métrica se puede usar para ver cualquier problema de rendimiento (latencia) con la canalización.
- Cantidad total de registros de datos leídos. Esta es una indicación general del número de registros emitidos por el conector. Este número debería aumentar constantemente, y reflejar la tendencia de escrituras en la base de datos subyacente de Spanner.
- Cantidad de particiones que se leen en este momento. Siempre se deben leer particiones. Si este número es cero, indica que se produjo un error en la canalización.
- Cantidad total de consultas realizadas durante la ejecución del conector. Esta es una indicación general de las consultas del flujo de cambios realizadas en la instancia de Spanner durante la ejecución de la canalización. Se puede usar para obtener una estimación de la carga del conector a la base de datos de Spanner.
Actualiza una canalización existente
Es posible actualizar una canalización en ejecución que usa el conector SpannerIO para procesar flujos de cambios si se aprueban las verificaciones de compatibilidad del trabajo. Para ello, debes establecer de forma explícita el parámetro de nombre de tabla de metadatos del trabajo nuevo cuando lo actualizas. Usa el valor de la opción de canalización metadataTable del trabajo que estás actualizando.
Si usas una plantilla de Dataflow proporcionada por Google, configura el nombre de la tabla con el parámetro spannerMetadataTableName. También puedes modificar tu trabajo existente para usar de forma explícita la tabla de metadatos con el método withMetadataTable(your-metadata-table-name) en la configuración del conector. Una vez hecho esto, puedes seguir las instrucciones en Inicia el trabajo de reemplazo de los documentos de Dataflow para actualizar un trabajo en ejecución.
Prácticas recomendadas para flujos de cambios y Dataflow
Las siguientes son algunas prácticas recomendadas para crear conexiones de flujos de cambios mediante Dataflow.
Usa una base de datos de metadatos independiente
Recomendamos crear una base de datos separada para que el conector SpannerIO la use para el almacenamiento de metadatos, en lugar de configurarla para usar la base de datos de tu aplicación.
Para obtener más información, consulta Considera usar una base de datos de metadatos independiente.
Ajusta el tamaño del clúster
Una regla general para una cantidad inicial de trabajadores en un trabajo de flujos de cambios de Spanner es un trabajador por cada 1,000 operaciones de escritura por segundo. Ten en cuenta que esta estimación puede variar según varios factores, como el tamaño de cada transacción, la cantidad de registros de flujos de cambios que se producen a partir de una sola transacción y otras transformaciones, agregaciones o receptores que se usan en la canalización.
Después de los recursos iniciales, es importante realizar un seguimiento de las métricas que se mencionan en Supervisa una canalización para garantizar que la canalización esté en buen estado. Recomendamos experimentar con un tamaño inicial del grupo de trabajadores y supervisar cómo tu canalización maneja la carga y aumentar la cantidad de nodos si es necesario. El uso de CPU es una métrica clave para verificar si la carga es adecuada y si se necesitan más nodos.
Limitaciones conocidas
Ajuste de escala automático
La compatibilidad con el ajuste de escala automático para cualquier canalización que incluya SpannerIO.readChangeStream requiere Apache Beam 2.39.0 o superior.
Si usas una versión de Apache Beam anterior a 2.39.0, las canalizaciones que incluyen SpannerIO.readChangeStream deben especificar de forma explícita el algoritmo de ajuste de escala automático como NONE, como se describe en Ajuste de escala automático horizontal.
Para escalar de forma manual una canalización de Dataflow en lugar de usar el ajuste de escala automático, consulta Escala de forma manual una canalización de transmisión.
Runner V2
El conector de flujos de cambios de Spanner requiere Dataflow Runner V2.
Se debe especificar manualmente durante la ejecución; de lo contrario, se mostrará un error. Puedes especificar Runner V2 si configuras tu trabajo con --experiments=use_unified_worker,use_runner_v2.
Instantánea
El conector de flujos de cambios de Spanner no es compatible con instantáneas de Dataflow.
Desviando
El conector de flujos de cambios de Spanner no admite el desvío de un trabajo. Solo se puede cancelar un trabajo existente.
También puedes actualizar una canalización existente sin necesidad de detenerla.
OpenCensus
Si quieres usar OpenCensus para supervisar tu canalización, especifica la versión 0.28.3 o una posterior.
NullPointerException cuando se inicia la canalización
Un error en la versión 2.38.0 de Apache Beam puede causar una NullPointerException cuando se inicia la canalización en ciertas condiciones. Esto evitaría que se inicie el trabajo y, en su lugar, se mostrará este mensaje de error:
java.lang.NullPointerException: null value in entry: Cloud Storage_PROJECT_ID=null
Para solucionar este problema, usa la versión 2.39.0 de Apache Beam o una posterior, o especifica de forma manual la versión de beam-sdks-java-core como 2.37.0:
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<version>2.37.0</version>
</dependency>