本页面演示了如何创建通过变更数据流使用和转发 Spanner 更改数据的 Dataflow 流水线。您可以使用此页面上的示例代码来构建自定义流水线。
核心概念
以下是用于变更数据流的 Dataflow 流水线的一些核心概念。
Dataflow
Dataflow 是一种快速且经济实惠的无服务器服务，支持流处理和批处理。它为使用开源 Apache Beam 库编写的处理作业提供了可移植性，并自动执行基础架构预配和集群管理。Dataflow 提供近乎实时的流式传输，从变更流读取数据时，延迟大约为 6 秒。
您可以通过 SpannerIO 连接器使用 Dataflow 来使用 Spanner 变更流，该连接器在 Spanner API 上提供了一个抽象化，用于查询变更数据流。借助此连接器，您无需管理变更数据流分区生命周期，但如果您直接使用 Spanner API，就需要管理变更数据流分区生命周期。该连接器可为您提供数据变更记录流，让您可以将更多精力放在应用逻辑上，而不必花费在特定的 API 详细信息和动态变更数据流分区上。在大多数情况下，如果您需要读取变更数据流数据，我们建议您使用 SpannerIO 连接器，而不是 Spanner API。
Dataflow 模板是实现常见用例的预构建 Dataflow 流水线。如需简要了解，请参阅 Dataflow 模板。
Dataflow 流水线
Spanner 变更数据流 Dataflow 流水线由四个主要部分组成：
- 包含变更数据流的 Spanner 数据库
- SpannerIO 连接器
- 用户定义的转换和接收器
- 接收器 I/O 写入器
下文将更详细地讨论以上各项。
Spanner 变更数据流
如需详细了解如何创建变更数据流，请参阅创建变更数据流。
Apache Beam SpannerIO 连接器
这是如前所述的 SpannerIO 连接器。它是一个源 I/O 连接器，可将数据更改历史记录的 PCollection 发出到流水线的后续阶段。每条发出的数据变更记录的事件时间将是提交时间戳。请注意，发出的记录无序，并且 SpannerIO 连接器保证没有延迟记录。
处理变更数据流时，Dataflow 会使用检查点。 因此，在缓冲更改时，每个工作器最多可能等待 5 秒，然后再发送更改以进一步处理。预计延迟时间约为 6 秒。
用户定义的转换
通过用户定义的转换，用户可以在 Dataflow 流水线中聚合、转换或修改处理数据。常见的使用场景包括移除个人身份信息、满足下游数据格式要求和排序。如需关于transforms的编程指南，请参阅 Apache Beam 官方文档。
Apache Beam 接收器 I/O 写入者
Apache Beam 包含内置 I/O 转换，该转换可用于从 Dataflow 流水线写入 BigQuery 等数据接收器。大多数常见的数据接收器都具有原生支持。
Dataflow 模板
借助 Dataflow 模板，您可以通过 Google Cloud 控制台、Google Cloud CLI 或 Rest API 调用，基于适用于常见使用场景的预构建 Docker 映像轻松创建 Dataflow 作业。
对于 Spanner 变更数据流，我们提供了三种 Dataflow 灵活模板：
构建 Dataflow 流水线
本部分介绍了连接器的初始配置，并提供了与 Spanner 变更数据流功能的常见集成示例。
如需执行以下步骤，您需要一个适用于 Dataflow 的 Java 开发环境。如需了解详情，请参阅使用 Java 创建 Dataflow 流水线。
创建变更流
如需详细了解如何创建变更流，请参阅创建变更流。 如需继续执行后续步骤，您必须拥有已配置变更数据流的 Spanner 数据库。
授予精细的访问权限控制权限
如果您希望任何精细的访问权限控制用户运行 Dataflow 作业，请确保用户有权访问一个数据库角色，该角色拥有变更数据流的 SELECT 权限和变更数据流的表值函数的 EXECUTE 权限。此外，请确保主帐号在 SpannerIO 配置或 Dataflow Flex 模板中指定了数据库角色。
如需了解详情，请参阅关于精细访问权限控制。
将 SpannerIO 连接器添加为依赖项
Apache Beam SpannerIO 连接器封装了通过 Cloud Spanner API 直接使用变更数据流的复杂性，可将由变更数据流数据记录的 PCollection 发送到流水线的后续阶段。
这些对象可在用户 Dataflow 流水线的其他阶段中使用。变更数据流集成是 SpannerIO 连接器的一部分。为了能够使用 SpannerIO 连接器，需要将依赖项添加到 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>
创建元数据数据库
运行 Apache Beam 流水线时，连接器需要跟踪每个分区。它会将此元数据保存在连接器在初始化期间创建的 Spanner 表中。配置连接器时，您可以指定要在其中创建此表的数据库。
如变更流最佳实践中所述，我们建议您为此新数据库创建新数据库，而不是允许连接器使用应用的数据库来存储其元数据表。
对于使用 SpannerIO 连接器的 Dataflow 作业，其所有者需要使用此元数据数据库设置以下 IAM 权限：
spanner.databases.updateDdlspanner.databases.beginReadOnlyTransactionspanner.databases.beginOrRollbackReadWriteTransactionspanner.databases.readspanner.databases.selectspanner.databases.writespanner.sessions.createspanner.sessions.get
配置连接器
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
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);
以下是 readChangeStream() 选项的说明：
Spanner 配置（必需）
用于配置在其中创建变更数据流并且应从其中查询变更数据流的项目、实例和数据库。 还可以选择指定当运行 Dataflow 作业的 IAM 主帐号是精细访问权限控制用户时要使用的数据库角色。该作业采用此数据库角色来访问变更数据流。如需了解详情，请参阅关于精细访问权限控制。
变更数据流名称（必填）
此名称唯一标识变更流。此处的名称必须与创建它时使用的名称相同。
元数据实例 ID（可选）
此实例用于存储连接器用于控制变更数据流 API 数据的使用的元数据。
元数据数据库 ID（必填）
该数据库用于存储连接器用来控制变更数据流 API 数据的使用的元数据。
元数据表名称（可选）
这只应在更新现有流水线时使用。
这是连接器要使用的现有元数据表名称。连接器使用此参数来存储元数据，以控制变更数据流 API 数据的使用量。如果省略此选项，Spanner 会在连接器初始化时创建一个具有生成名称的新表。
RPC 优先级（可选）
要用于变更数据流查询的请求优先级。如果省略此参数，将使用 high
priority。
InclusiveStartAt（必需）
系统会将给定时间戳的更改返回给调用方。
InclusiveEndAt（可选）
在指定时间戳之前的更改将返回给调用方。如果省略此参数，系统会无限期地发出更改。
添加转换和接收器以处理更改数据
完成上述步骤后，配置的 SpannerIO 连接器已准备好发出由 DataChangeRecord 对象的 PCollection。如需查看以各种方式处理此流式数据的几个示例流水线配置，请参阅转换和接收器示例。
请注意，SpannerIO 连接器发出的变更数据流记录是无序的。这是因为 PCollection 不提供任何排序保证。如果您需要有序的数据流，则必须在流水线中以转换的形式对记录进行分组和排序：请参阅示例：按键排序。您可以扩展此示例，以根据记录的任何字段（例如按事务 ID）对记录进行排序。
转换和接收器示例
您可以定义自己的转换并指定将数据写入的接收器。Apache Beam 文档提供了大量可以应用的transforms，并且您现在可直接使用 I/O 连接器将数据写入外部系统。
示例：按键排序
此代码示例使用 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
此代码示例使用状态和计时器来缓冲每个键的记录，并将计时器的到期时间设置为用户配置的某个未来时间 T（在 BufferKeyUntilOutputTimestamp 函数中定义）。当 Dataflow 水印经过时间 T 时，此代码会清空缓冲区中时间戳小于 T 的所有记录，按提交时间戳对这些记录进行排序，并输出键值对，其中：
- 该键为输入键，即，主键已哈希处理为大小为 1000 的存储桶数组的主键。
- 该值是为键缓冲的有序数据变更记录。
对于每个密钥，我们提供以下保证：
- 计时器保证会按照到期时间戳触发。
- 下游阶段保证按生成顺序接收元素。
例如，假设对于值为 100 的键，计时器分别在 T1 和 T10 触发，在每个时间戳都生成一组数据变更记录。在 T1 输出的数据更改记录是在 T10 输出的数据更改记录之前生成的，因此在 T10 输出的数据更改记录之前，在 T1 输出的数据更改记录也一定会被下一阶段收到。这种机制有助于我们确保每个主键的提交时间戳严格排序，以便进行下游处理。
此过程将重复进行，直到流水线结束且所有数据更改记录均处理完毕（或者，如果未指定结束时间，则将无限期地重复）。
请注意，此代码示例使用状态和计时器（而不是窗口）按键执行排序。原因在于窗口不一定会按顺序处理。这意味着，较旧窗口的处理时间晚于较新窗口的处理时间，这可能会导致处理顺序混乱。
BreakRecordByModFn
每条数据更改记录都可以包含多个 mod。每个 mod 表示对单个主键值执行插入、更新或删除操作。此函数将每条数据更改记录拆分为单独的数据更改记录，每个 mod 对应一个数据更改记录。
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
此函数接受一个 DataChangeRecord，并输出一个由 Spanner 主键键控且已转换为整数值的 DataChangeRecord。
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
计时器和缓冲区是按键进行的。此函数会缓冲每个数据更改记录，直到水印超过我们想要输出已缓冲数据更改记录的时间戳为止。
此代码利用循环计时器来确定何时刷新缓冲区：
- 当它第一次看到某个键的数据更改记录时，它会将计时器设置为在数据更改记录的提交时间戳 +
incrementIntervalSeconds（用户可配置的选项）时触发。 - 当计时器触发时，它会将缓冲区中时间戳小于计时器到期时间的所有数据更改记录添加到
recordsToOutput。如果缓冲区具有时间戳大于或等于计时器到期时间的数据变更记录，则将数据更改记录添加回缓冲区，而不是输出。然后将下一个计时器设置为当前计时器的到期时间加上incrementIntervalInSeconds。 - 如果
recordsToOutput不为空，该函数会按提交时间戳和事务 ID 对recordsToOutput中的数据更改记录进行排序，然后输出这些记录。
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();
}
}
}
对交易进行排序
可以按事务 ID 和提交时间戳更改此流水线。为此，请缓冲每个事务 ID / 提交时间戳对（而不是每个 Spanner 键）的记录。这需要修改 KeyByIdFn 中的代码。
示例：汇编事务
此代码示例可读取数据更改记录，将属于同一事务的所有数据更改记录组合到单个元素中，然后输出该元素。请注意，此示例代码输出的事务未按提交时间戳排序。
此代码示例使用缓冲区来根据数据更改记录组建事务。在首次收到属于某事务的数据更改记录时，它会读取数据更改记录中的 numberOfRecordsInTransaction 字段，该字段用于描述属于该事务的预期数据更改记录数量。它会缓冲属于该事务的数据变更记录，直到缓冲记录的数量与 numberOfRecordsInTransaction 匹配，随后输出捆绑的数据变更记录。
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
此函数接受 DataChangeRecord 并输出由交易 ID 键控的 DataChangeRecord。
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
TransactionBoundaryFn 缓冲区从 KeyByTransactionIdFn 接收 {TransactionId, DataChangeRecord} 的键值对，并根据 TransactionId 将它们分组。当缓冲的记录数等于整个事务中包含的记录数时，此函数会按记录序列对组中的 DataChangeRecord 对象进行排序，并输出 {CommitTimestamp, TransactionId} 和 Iterable<DataChangeRecord> 的键值对。
此处，我们假设 SortKey 是表示 {CommitTimestamp, TransactionId} 对的用户定义类。请参阅 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();
}
}
}
示例：按交易代码过滤
为修改用户数据的交易添加代码后，相应的代码及其类型会存储为 DataChangeRecord 的一部分。以下示例演示了如何根据用户指定的交易标记以及系统标记过滤变更数据流记录：
针对 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
系统标记过滤/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
示例：提取完整行
此示例使用名为 Singer 且具有以下定义的 Spanner 表：
CREATE TABLE Singers (
SingerId INT64 NOT NULL,
FirstName STRING(1024),
LastName STRING(1024)
) PRIMARY KEY (SingerId);
在变更数据流的默认 OLD_AND_NEW_VALUES 值捕获模式下，当 Spanner 行有更新时，收到的数据更改记录将仅包含已更改的列。跟踪但未更改的列将不会包含在记录中。mod 的主键可用于在数据更改记录的提交时间戳执行 Spanner 快照读取，以提取未更改的列，甚至检索整行。
请注意，数据库保留政策可能需要更改为大于或等于变更数据流保留政策的值，才能成功读取快照。
另请注意，建议使用 NEW_ROW 值捕获类型来执行此操作，因为该方法默认返回该行的所有跟踪列，并且不需要将额外的快照读入 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
此转换将在收到的每条记录的提交时间戳处执行过时读取，并将整行映射到 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());
});
}
}
这段代码会创建一个 Spanner 数据库客户端以执行完整行提取，并将会话池配置为仅包含几个会话，从而依序在 ToFullRowJsonFn 的一个实例中执行读取操作。Dataflow 会确保生成此函数的多个实例，每个实例都有自己的客户端池。
示例：Spanner 到 Pub/Sub
在这种情况下，调用方会尽快将记录流式传输到 Pub/Sub，而不进行任何分组或汇总。这非常适合触发下游处理，例如将插入 Spanner 表的所有新行流式传输到 Pub/Sub 以供进一步处理。
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"));
请注意，您可以将 Pub/Sub 接收器配置为确保“正好一次”语义。
示例：Spanner 到 Cloud Storage
在这种情况下，调用方会对给定窗口中的所有记录进行分组，并将记录组保存在单独的 Cloud Storage 文件中。这种做法非常适合用于分析和时间点归档，后者独立于 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));
请注意，Cloud Storage 接收器默认提供“至少一次”语义。通过额外的处理，可将其修改为具有“正好一次”语义。
我们还为此使用场景提供了 Dataflow 模板：请参阅将变更数据流连接到 Cloud Storage。
示例：Spanner 到 BigQuery（分类记录表）
在这里，调用方将变更记录流式传输到 BigQuery 中。每条数据更改记录在 BigQuery 中显示为一行。非常适合用于分析。此代码使用之前在提取完整行部分中定义的函数来检索记录的完整行并将其写入 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());
}
)
);
请注意，BigQuery 接收器默认提供“至少一次”语义。通过额外的处理，可将其修改为具有“正好一次”语义。
我们还为此用例提供了 Dataflow 模板；请参阅将变更数据流连接到 BigQuery。
监控流水线
有两类指标可用于监控变更数据流 Dataflow 流水线。
标准 Dataflow 指标
Dataflow 提供了多个指标来确保作业健康状况良好，例如数据新鲜度、系统延迟、作业吞吐量、工作器 CPU 利用率等。如需了解详情，请参阅对 Dataflow 流水线使用 Monitoring。
对于变更数据流流水线，您需要考虑两个主要指标：系统延迟时间和数据新鲜度。
系统延迟会告知您已处理或等待处理某个数据项的当前最长时长（以秒为单位）。
数据新鲜度将显示从现在（实时）到输出水印之间的时间。时间 T 的输出水印表示事件时间（严格）早于 T 的所有元素均已处理完毕以进行计算。换言之，数据新鲜度用于衡量流水线在处理收到的事件方面的最新程度。
如果流水线资源不足，您可以通过这两个指标了解这种影响。系统延迟时间会增加，因为内容需要等待更长时间才能得到处理。数据新鲜度也会提高，因为流水线无法跟上收到的数据量。
自定义变更数据流指标
这些指标在 Cloud Monitoring 中提供，包括：
- 从在 Spanner 中提交记录到连接器将其发送到 PCollection 之间的分桶（直方图）延迟时间。此指标可用于查看流水线的任何性能（延迟）问题。
- 读取的数据记录总数。这是连接器发出的记录数的总体指示。此数字应该在不断增加，与底层 Spanner 数据库中的写入趋势相符。
- 当前读取的分区数。应该始终有读取的分区。如果此数字为零，则表示流水线中出现了错误。
- 在连接器执行期间发出的查询总数。这是整体指示在流水线执行过程中对 Spanner 实例进行的变更数据流查询。这可用于估算从连接器到 Spanner 数据库的负载。
更新现有流水线
如果作业兼容性检查通过，可以更新使用 SpannerIO 连接器处理变更数据流的正在运行的流水线。为此，您必须在更新新作业时明确设置新作业的元数据表名称参数。使用要更新的作业中的 metadataTable 流水线选项的值。
如果您使用的是 Google 提供的 Dataflow 模板，请使用参数 spannerMetadataTableName 设置表名称。您还可以修改现有作业，通过连接器配置中的 withMetadataTable(your-metadata-table-name) 方法明确使用元数据表。完成此操作后，您可以按照 Dataflow 文档中启动替换作业中的说明来更新正在运行的作业。
变更数据流和 Dataflow 的最佳实践
以下是使用 Dataflow 构建变更数据流连接的一些最佳实践。
使用单独的元数据数据库
我们建议为 SpannerIO 连接器创建用于元数据存储的单独数据库，而不是将其配置为使用您的应用数据库。
如需了解详情，请参阅考虑使用单独的元数据数据库。
调整集群大小
对于 Spanner 变更数据流作业中初始工作器数量，一般而言，每秒每 1,000 次写入对应 1 个工作器。请注意，此估算值可能会因多个因素而异，例如每个事务的大小、单个事务生成的变更数据流记录数量，以及流水线中使用的其他转换、汇总或接收器数量。
完成初始资源分配后，请务必跟踪监控流水线中提到的指标，以确保流水线正常运行。我们建议您尝试使用初始工作器池大小，并监控流水线如何处理负载，以便在必要时增加节点数量。CPU 利用率是检查负载是否适当以及是否需要更多节点的关键指标。
已知限制
自动扩缩
若要支持任何包含 SpannerIO.readChangeStream 的流水线，您需要具备 Apache Beam 2.39.0 或更高版本。
如果您使用的是 2.39.0 之前的 Apache Beam 版本，则包含 SpannerIO.readChangeStream 的流水线需要将自动扩缩算法明确指定为 NONE，如横向自动扩缩中所述。
如需手动伸缩 Dataflow 流水线而不是使用自动伸缩，请参阅手动伸缩流处理流水线。
Runner V2
Spanner 变更数据流连接器需要 Dataflow Runner V2。必须在执行期间手动指定，否则系统会抛出错误。您可以通过使用 --experiments=use_unified_worker,use_runner_v2 配置作业来指定 Runner V2。
快照
Spanner 变更数据流连接器不支持 Dataflow 快照。
正在排空
Spanner 变更数据流连接器不支持排空作业。只能取消现有作业。
您还可以更新现有流水线，而无需停止该流水线。
OpenCensus
如需使用 OpenCensus 监控流水线，请指定 0.28.3 或更高版本。
流水线启动时 NullPointerException
在特定条件下启动流水线时，Apache Beam 版本 2.38.0 中的一个 bug 可能会导致 NullPointerException。这会导致您的作业无法启动，并改为显示以下错误消息：
java.lang.NullPointerException: null value in entry: Cloud Storage_PROJECT_ID=null
如需解决此问题，请使用 Apache Beam 版本 2.39.0 或更高版本，或手动将 beam-sdks-java-core 版本指定为 2.37.0：
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<version>2.37.0</version>
</dependency>