实时写入数据
RealtimeStream
RealtimeStream
是 Java SDK 的实时写入接口，适合把 Java 应用中的事件、日志、指标或 CDC 数据持续写入 Lakehouse 表。它面向高频小批量写入，数据写入后可秒级查询。
概述
实时写入的数据会先进入 Lakehouse Ingestion Service。普通查询可以较快看到新写入数据；Table Stream、物化视图和动态表只能读取已提交数据，通常需要等待约 1 分钟。
| 写入模式 | 适用表 | 支持操作 | 典型场景 |
|---|
APPEND_ONLY
APPEND_ONLY | 普通表 | INSERT
INSERT | 日志、埋点、指标等追加写入 |
CDC
CDC | 主键表 | INSERT
INSERT 、UPDATE
UPDATE 、UPSERT
UPSERT 、DELETE
DELETE 等 | 数据库 CDC、主键维表变更同步 |
使用限制
- 表结构变更时，需要先停止实时写入任务；变更完成后等待一段时间再重新启动写入任务。
- Table Stream、物化视图和动态表只能读取已提交数据，不能立即读取实时写入的未提交数据。
APPEND_ONLY
APPEND_ONLY
模式只能创建 Stream.Operator.INSERT
Stream.Operator.INSERT
类型的 Row。CDC
CDC
模式建议写入定义了主键的表，否则无法稳定表达更新和删除语义。
创建追加写入流
import com.clickzetta.client.ClickZettaClient;
import com.clickzetta.client.RealtimeStream;
import com.clickzetta.client.RowStream;
import com.clickzetta.platform.client.api.FlushMode;
import com.clickzetta.platform.client.api.Options;
import com.clickzetta.platform.client.api.Row;
import com.clickzetta.platform.client.api.Stream;
public class RealtimeAppendDemo {
public static void main(String[] args) throws Exception {
String jdbcUrl = args[0];
String username = args[1];
String password = args[2];
String schema = args[3];
String table = args[4];
ClickZettaClient client = ClickZettaClient.newBuilder()
.url(jdbcUrl)
.username(username)
.password(password)
.build();
Options options = Options.builder()
.withFlushMode(FlushMode.AUTO_FLUSH_BACKGROUND)
.withMutationBufferLinesNum(1000)
.withFlushInterval(10_000)
.build();
RealtimeStream stream = client.newRealtimeStreamBuilder()
.operate(RowStream.RealTimeOperate.APPEND_ONLY)
.options(options)
.schema(schema)
.table(table)
.build();
try {
Row row = stream.createRow(Stream.Operator.INSERT);
row.setValue("id", 1);
row.setValue("name", "event_001");
stream.apply(row);
stream.flush();
} finally {
stream.close();
client.close();
}
}
}
创建 CDC 写入流
CDC 模式用于主键表的变更写入，常用操作包括
UPSERT
UPSERT
和
DELETE_IGNORE
DELETE_IGNORE
。
import com.clickzetta.client.ClickZettaClient;
import com.clickzetta.client.RealtimeStream;
import com.clickzetta.client.RowStream;
import com.clickzetta.platform.client.api.Options;
import com.clickzetta.platform.client.api.Row;
import com.clickzetta.platform.client.api.Stream;
public class RealtimeCdcDemo {
public static void main(String[] args) throws Exception {
String jdbcUrl = args[0];
String username = args[1];
String password = args[2];
String schema = args[3];
String table = args[4];
ClickZettaClient client = ClickZettaClient.newBuilder()
.url(jdbcUrl)
.username(username)
.password(password)
.build();
RealtimeStream stream = client.newRealtimeStreamBuilder()
.operate(RowStream.RealTimeOperate.CDC)
.options(Options.builder().build())
.schema(schema)
.table(table)
.build();
try {
Row upsertRow = stream.createRow(Stream.Operator.UPSERT);
upsertRow.setValue("id", 1);
upsertRow.setValue("name", "updated_name");
stream.apply(upsertRow);
Row deleteRow = stream.createRow(Stream.Operator.DELETE_IGNORE);
deleteRow.setValue("id", 2);
stream.apply(deleteRow);
stream.flush();
} finally {
stream.close();
client.close();
}
}
}
Options 参数
Options
Options
用于控制实时写入的缓冲、刷新和失败重试。多数场景可以使用默认值，仅在吞吐量或延迟不满足预期时调整。
| 参数 | 默认值 | 说明 |
|---|
withFlushMode
withFlushMode | FlushMode.AUTO_FLUSH_BACKGROUND
FlushMode.AUTO_FLUSH_BACKGROUND | 刷写模式；异步刷写吞吐更高，同步刷写更容易控制顺序 |
withMutationBufferLinesNum
withMutationBufferLinesNum | 1000
1000 | 单个 buffer 累积的最大行数，达到后触发 flush |
withMutationBufferSpace
withMutationBufferSpace | 10 * 1024 * 1024
10 * 1024 * 1024 | 单个 buffer 累积的最大字节数，达到后触发 flush |
withMutationBufferMaxNum
withMutationBufferMaxNum | 10
10 | 同时存在的最大 buffer 数量，用于控制并发和内存占用 |
withFlushInterval
withFlushInterval | 10 * 1000
10 * 1000 | 定时 flush 间隔，单位毫秒 |
withRequestFailedRetryEnable
withRequestFailedRetryEnable | true
true | 是否开启失败重试 |
withRequestFailedRetryTimes
withRequestFailedRetryTimes | 5
5 | 失败后最大重试次数 |
withRequestFailedRetryInternalMs
withRequestFailedRetryInternalMs | 5000
5000 | 重试间隔，单位毫秒 |
flush 和 close
stream.apply(row)
stream.apply(row)
把 Row 写入 SDK 缓冲区。stream.flush()
stream.flush()
主动把缓冲数据发送到服务端。stream.close()
stream.close()
会释放流资源，关闭前会隐式 flush。
⚠️ 注意：生产任务中必须确保异常路径也能执行
close()
close()
，否则可能出现缓冲数据未提交或资源未释放。
相关文档
