云妹导读：node
前不久，京东智联云正式上线了基于Clickhouse的分析型云数据库JCHDB，一经推出便受到广大用户的极大关注。有兴趣的小伙伴能够回顾上一篇文章**《比MySQL快839倍！揭开分析型数据库JCHDB的神秘面纱》**。数据库
ClickHouse像ElasticSearch同样具备数据分片（shard）的概念，这也是分布式存储的特色之一，即经过并行读写提升效率。ClickHouse依靠Distributed引擎实现了Distributed（分布式）表机制，在全部分片（本地表）上创建视图进行分布式查询，使用很方便。express
Distributed表引擎是**一种特殊的表引擎，自身不会存储任何数据，而是经过读取或写入其余远端节点上的表进行数据处理的表引擎。**该表引擎须要依赖各个节点的本地表来建立，本地表的存在是Distributed表建立的依赖条件，建立语句以下：异步
CREATE TABLE {teble} ON CLUSTER {cluster}
AS {local_table}
ENGINE= Distributed({cluster}, {database}, {local_table}，{policy})
这里的policy通常可使用随机（例如rand()）或哈希（例如halfMD5hash(id)）。tcp
再来看下ClickHouse集群节点配置文件，相关参数以下：分布式
<remote_servers>
<logs>
<shard>
<weight>1</weight>
<internal_replication>true</internal_replication>
<replica>
<priority>1</priority>
<host>example01-01-1</host>
<port>9000</port>
</replica>
<replica>
<host>example01-01-2</host>
<port>9000</port>
</replica>
</shard>
<shard>
<weight>2</weight>
<internal_replication>true</internal_replication>
<replica>
<host>example01-02-1</host>
<port>9000</port>
</replica>
<replica>
<host>example01-02-2</host>
<port>9000</port>
</replica>
</shard>
</logs>
</remote_servers>
有了上面的基础了解，就将进入主题了，本文主要是对Distributed表如何写入及如何分发作一下分析，略过SQL的词法解析、语法解析等步骤，从写入流开始，其构造方法以下：函数
DistributedBlockOutputStream(const Context & context_, StorageDistributed & storage_, const ASTPtr & query_ast_, const ClusterPtr & cluster_, bool insert_sync_, UInt64 insert_timeout_);
若是insert_sync_为true，表示是同步写入，并配合insert_timeout_参数使用（insert_timeout_为零表示没有超时时间）；若是insert_sync_为false，表示写入是异步。优化
1，同步写入仍是异步写入ui
同步写入是指数据直写入实际的表中，而异步写入是指数据首先被写入本地文件系统，而后发送到远端节点。this
BlockOutputStreamPtr StorageDistributed::write(const ASTPtr &, const Context &
context)
{
......
/// Force sync insertion if it is remote() table function
bool insert_sync = settings.insert_distributed_sync || owned_cluster;
auto timeout = settings.insert_distributed_timeout;
/// DistributedBlockOutputStream will not own cluster, but will own
ConnectionPools of the cluster
return std::make_shared(
context, *this, createInsertToRemoteTableQuery(remote_database,
remote_table, getSampleBlockNonMaterialized()), cluster,
nsert_sync, timeout);
}
是否执行同步写入是由insert_sync决定的，最终是由是否配置insert_distributed_sync（默认为false）和owned_cluster值的或关系决定的，通常在使用MergeTree之类的普通表引擎时，一般是异步写入，但在使用表函数时（使用owned_cluster来判断是不是表函数），一般会使用同步写入。这也是在设计业务逻辑时须要注意的。
owned_cluster是何时赋值的呢？
StoragePtr TableFunctionRemoteexecuteImpl(const ASTPtr & astfunction, const Context &
context, const stdstring & tablename) const
{
......
StoragePtr res = remotetablefunction_ptr
? StorageDistributed::createWithOwnCluster(
table_name,
structureremotetable,
remotetablefunction_ptr,
cluster,
context)
: StorageDistributed::createWithOwnCluster(
table_name,
structureremotetable,
remote_database,
remote_table,
cluster,
context);
......
}
StoragePtr StorageDistributed::createWithOwnCluster(
const std::string & tablename,
const ColumnsDescription & columns_,
ASTPtr & remotetablefunctionptr,
ClusterPtr & ownedcluster,
const Context & context_)
{
auto res = create(String{}, tablename, columns, ConstraintsDescription{},
remotetablefunctionptr, String{}, context, ASTPtr(), String(), false);
res->ownedcluster = ownedcluster_;
return res;
}
能够发如今建立remote表时会根据remote_table_function_ptr参数对最终的owned_cluster_赋值为true。
2，异步写入是如何实现的
了解了何时使用同步写入何时异步写入后，再继续分析正式的写入过程，同步写入通常场景中涉及较少，这里主要对异步写入逻辑进行分析。outStream的write方法主逻辑以下：
DistributedBlockOutputStream::write()
↓
if insert_sync
| |
true false
↓ ↓
writeSync() writeAsync()
其实这个write方法是重写了virtual void IBlockOutputStream::write(const Block & block)，因此节点在接收到流并调用流的write方法就会进入该逻辑中。而且根据insert_sync来决定走同步写仍是异步写。
3，写入本地节点仍是远端节点
主要仍是对异步写入进行分析，其实writeAsync()最终的实现方法是writeAsyncImpl()，大体逻辑图以下：
writeAsyncImpl()
↓
if shard_info.hasInternalReplication()
| |
true false
↓ ↓
writeToLocal() writeToLocal()
↓ ↓
writeToShard() for(every shard){writeToShard()}
↓ ↓
end end
其中getShardsInfo()方法就是获取config.xml配置文件中获取集群节点信息，hasInternalReplication()就对应着配置文件中的internal_replication参数，若是为true，就会进入最外层的if逻辑，不然就会进入else逻辑。
其中writeToLocal()方法是相同的，是指若是shard包含本地节点，优先选择本地节点进行写入；后半部分writeToShard()就是根据internal_replication参数的取值来决定是写入其中一个远端节点，仍是全部远端节点都写一次。
4，数据如何写入本地节点
固然通常状况Distributed表仍是基于ReplicatedMergeTree系列表进行建立，而不是基于表函数的，因此大多数场景仍是会先写入本地再分发到远端节点。那写入Distributed表的数据是如何保证原子性落盘而不会在数据正在写入的过程当中就把不完整的数据发送给远端其余节点呢？看下writeToShard()方法大体逻辑，以下：
writeToShard()
↓
for(every dir_names){
|
└──if first iteration
| |
false true
↓ ↓
| ├──storage.requireDirectoryMonitor()
| ├──CompressedWriteBuffer
| ├──writeStringBinary()
| ├──stream.writePrefix()
| ├──stream.write(block)
| ├──stream.writeSuffix()
↘ ↙
link(tmp_file, file)
└──}
继续具体再看下源码的具体实现，以下：
void DistributedBlockOutputStream::writeToShard(const Block & block, const
std::vector<std::string> & dir_names)
{
/** tmp directory is used to ensure atomicity of transactions
* and keep monitor thread out from reading incomplete data
*/
std::string first_file_tmp_path{};
auto first = true;
/// write first file, hardlink the others
for (const auto & dir_name : dir_names)
{
const auto & path = storage.getPath() + dir_name + '/';
/// ensure shard subdirectory creation and notify storage
if (Poco::File(path).createDirectory())
storage.requireDirectoryMonitor(dir_name);
const auto & file_name = toString(storage.file_names_increment.get()) +
".bin";
const auto & block_file_path = path + file_name;
/** on first iteration write block to a temporary directory for
subsequent hardlinking to ensure
* the inode is not freed until we're done */
if (first)
{
first = false;
const auto & tmp_path = path + "tmp/";
Poco::File(tmp_path).createDirectory();
const auto & block_file_tmp_path = tmp_path + file_name;
first_file_tmp_path = block_file_tmp_path;
WriteBufferFromFile out{block_file_tmp_path};
CompressedWriteBuffer compress{out};
NativeBlockOutputStream stream{compress, ClickHouseRevision::get(),
block.cloneEmpty()};
writeVarUInt(UInt64(DBMS_DISTRIBUTED_SENDS_MAGIC_NUMBER), out);
context.getSettingsRef().serialize(out);
writeStringBinary(query_string, out);
stream.writePrefix();
stream.write(block);
stream.writeSuffix();
}
if (link(first_file_tmp_path.data(), block_file_path.data()))
throwFromErrnoWithPath("Could not link " + block_file_path + " to "
+ first_file_tmp_path, block_file_path,
ErrorCodes::CANNOT_LINK);
}
......
}
首先来了解下Distributed表在目录中的存储方式，默认位置都是/var/lib/clickhouse/data/{database}/{table}/在该目录下会为每一个shard生成不一样的目录，其中存放须要发送给该shard的数据文件，例如：
[root@ck test]# tree . ├── 'default@ck2-0:9000,default@ck2-1:9000' │ ├── 25.bin │ └── tmp │ └── 26.bin └── 'default@ck3-0:9000,default@ck3-1:9000' └── tmp
能够发现每一个shard对应的目录名是{darabse}@{hostname}:{tcpPort}的格式，若是多个副本会用，分隔。而且每一个shard目录中还有个tmp目录，这个目录的设计在writeToShard()方法中作了解释，是为了不数据文件在没写完就被发送到远端。
数据文件在本地写入的过程当中会先写入tmp路径中，写完后经过硬连接link到shard目录，保证只要在shard目录中出现的数据文件都是完整写入的数据文件。
数据文件的命名是经过全局递增的数字加.bin命名，是为了在后续分发到远端节点保持顺序性。
5，数据如何分发到各个节点
细心的你可能已经发如今writeToShard()方法中有个requireDirectoryMonitor()，这个方法就是将shard目录注册监听，并经过专用类StorageDistributedDirectoryMonitor来实现数据文件的分发，根据不一样配置能够实现逐一分发或批量分发。而且包含对坏文件的容错处理。
分析到这，可能还有人会以为云里雾里，以为整个流程串不起来，其实这样写是为了先不影响Distributed表写入的主流程，明白了这个再附加上sharding_key拆分和权重拆分就很好理解了。
上面提到过writeAsync()的最终实现方法是writeAsyncImpl，这个说法是没问题的，可是中间还有段关键逻辑，以下：
writeAsync()
↓
if storage.getShardingKeyExpr() && (cluster->getShardsInfo().size() > 1
| |
true false
↓ ↓
writeAsyncImpl(block) writeSplitAsync(block)
↓
splitBlock(block)
↓
writeAsyncImpl(splitted_blocks,shard_idx)
getShardingKeyExpr()方法就是去获取sharding_key生成的表达式指针，该表达式是在建立表时就生成的，以下：
sharding_key_expr = buildShardingKeyExpression(sharding_key_, global_context, getColumns().getAllPhysical(), false);
那sharding_key和sharding_key_expr是什么关系呢？以下：
const ExpressionActionsPtr & getShardingKeyExpr() const { return
sharding_key_expr; }
因此说sharding_key_expr最终主要就是由sharding_key决定的。
通常状况下getShardingKeyExpr()方法都为true，若是再知足shard数量大于1，就会对block进行拆分，由splitBlock()方法主要逻辑就是建立selector并使用selector进行切割，大体逻辑以下：
splitBlock()
↓
createSelector(block)
↓
for(every shard){column->scatter(num_shards, selector);}
对于如何建立selector以及selector中都作了什么事儿，来具体看下源码截取，以下：
IColumn::Selector DistributedBlockOutputStream::createSelector(const Block &
source_block)
{
Block current_block_with_sharding_key_expr = source_block;
storage.getShardingKeyExpr()-
>execute(current_block_with_sharding_key_expr);
const auto & key_column =
current_block_with_sharding_key_expr.getByName(storage.getShardingKeyColumnName
());
const auto & slot_to_shard = cluster->getSlotToShard();
......
throw Exception{"Sharding key expression does not evaluate to an integer
type", ErrorCodes::TYPE_MISMATCH};
}
看splitBlock()方法，ClickHouse是利用createSelector()方法构造selector来进行后续的处理。在createSelector()方法中最重要的就是key_column和slot_to_shard。
key_column是经过sharding_key间接得到的，是为了根据主键列进行切割；slot_to_shard是shard插槽，这里就是为了处理权重，在后续向插槽中插入数据时就会结合config.xml中的weight进行按比例处理。
细节比较复杂这里不作太细致的分析，有兴趣能够自行看下（如template IColumn::Selector createBlockSelector()）。
到此，对于Distributed表的写入流程的关键点就大体分析完了。篇幅有限有些细节没有作过多说明，有兴趣的能够自行再了解下。
经过对Distributed表写入流程的分析，了解了该类型表的实际工做原理，因此在实际应用中有几个点还须要关注一下：
- Distributed表在写入时会在本地节点生成临时数据，会产生写放大，因此会对CPU及内存形成一些额外消耗，建议尽可能少使用Distributed表进行写操做；
- Distributed表写的临时block会把原始block根据sharding_key和weight进行再次拆分，会产生更多的block分发到远端节点，也增长了merge的负担；
- Distributed表若是是基于表函数建立的，通常是同步写，须要注意。
了解原理才能更好的使用，遇到问题才能更好的优化。
点击【阅读原文】便可前往京东智联云控制台开通试用JCHDB。