ByteHouse 是一款分布式数据库，其默认行为不考虑读写一致性，副本间的数据在一段时间后会达到最终一致。本文将介绍引起 ByteHouse 读写不一致的因素以及在数据加工等场景下的最佳实践配置，以帮助用户在使用 ByteHouse 时更好地处理读写一致性问题。

ByteHouse 的默认写入行为是异步写入，数据在写入分布式表后，会根据分片键拆分并在分片中找到合适的副本（通常是 Leader 副本）进行异步写入，然后该副本会将数据同步到同分片的另一副本。
读取分布式表时，分布式表会找到每个分片中符合要求的副本（数据延迟最大 120s，可通过用户参数max_replica_delay_for_distributed_queries调整）。因此，写入 ByteHouse 集群的数据不能立即被读到，有以下常见原因：

• 分布式写入延迟：数据积压在写入的分布式表，因集群负载因素，异步写入到 Local 表非常缓慢。
• 主备延迟：输入成功写入了一个节点的 Local 表，因集群负载因素，同步到同分片异副本的 Local 表非常缓慢。（如果单副本时，不存在此因素）
• 其他原因。如：
  • 分布式 DDL 延迟：部分分布式 DDL 超时导致数据不一致。
  • 写入延迟：默认攒批时间长导致的数据延迟。

在数据加工等场景下，如果需要写入的数据立即被读到，可以根据以下指导进行设置。但由于本身不是 ByteHouse 的默认行为，也注意如此操作的风险。

避免分布式异步插入导致不一致

将默认的异步行为改为同步，这样等待插入 SQL 会在全部副本插入成功后才会返回。

<!-- 用户参数 -->
<insert_distributed_sync>true<insert_distributed_sync>

注意：此操作会导致插入返回变慢，且插入不会攒批执行，可能加重 Merge 负担。

避免副本同步导致的不一致

单副本

彻底避免副本同步导致的不一致，最简单的方案即是使用单副本。
当然单副本方案，会在集群遭遇单节点故障时不可用，需要权衡可用性、数据可靠性与一致性，再决策是否使用单副本。

双副本

双副本下，不同引擎的行为不一致，请根据选用的引擎调整使用方式。

HaMergeTree

HaMergeTree 没有 Leader 和非 Leader 之分，通过设置让插入与查询时均以顺序选择节点，即可做到写入的数据立刻可读。

<!-- 用户参数 -->
<load_balancing>in_order</load_balancing>

但注意，该操作的代价是另一份副本无法被发送到查询，仅作为备节点的存在，CPU和内存都会被浪费。建议仅核对数据核对的场景才指定in_order查询，其他情况下还是以默认值random查询。

HaUniqueMergeTree

为了进行去重 ，HaUniqueMergeTree 有表级的 leader 与非 Leader，仅 Leader 会写入和去重数据。因此，可以通过建表时增加 Settings，让读取分布式表时永远只读取 Leader 表；

Create Table ...
Engine=HaUniqueTable ...
Settings ha_unique_read_session_consistency = true;

避免 DDL 导致的不一致

对某节点发送分布式 DDL（on cluster）进行 Create，Alter，Drop，Truncate 时，该节点会等待所有节点的 DDL 均执行成功才返回成功。但是分布式 DDL 有超时时间，超时后仍会继续执行。
如果在数据链路中，有场景需要等待上一步建表、清理数据等，建议取消超时时间：

<!-- 用户参数 -->
<distributed_ddl_task_timeout>0</distributed_ddl_task_timeout>

注意：分布式DDL是一个队列，在集群繁忙时，该操作可能花费较长时间。

避免插入导致的不一致

Insert Into 插入

ByteHouse 的 Insert Into 语法保持了原子性，不管插入多少数据，全部插入成功才会返回成功，任意一条数据插入失败就会返回失败，且只要写入成功，数据会立即落盘，在当前节点就立即可以查到。
因此，任何情况下都保证插入有重试机制，即可做到写入的数据立刻被读到。

Kafka

除了性能原因导致上游数据来不及消费，Kafka 消费后数据不立即可见的原因是后台的攒批机制。
若希望减少数据延迟，可以减少攒批时间（默认为8000，即8秒）；

ALTER TABLE db.kafka_table MODIFY SETTING stream_flush_interval_ms = 1000;

但建议仍至少保持一秒，并请关注inode等监控指标，避免因为文件过碎导致系统的 Merge 负担增大。

Flink

和 Kafka 类似，希望减少数据延迟，也可以通过设置减少攒批时间；
与 Kafka 的建议也一致：至少保持一秒；

CREATE TABLE lookup_table (
  ...
) WITH (
  'connector' = 'bytehouse-ce',
  ...
  'sink.buffer-flush.interval' = '1 second'
);