【ClickHouse源码】Distributed之表insert流程

Bertha 。 2022-12-22 09:28 180阅读 0赞

### Distributed表引擎介绍 ###

`Distributed`表引擎是一种特殊的表引擎，自身不会存储任何数据，而是通过读取或写入其他远端节点上的表进行数据处理的表引擎。该表引擎需要依赖各个节点的本地表来创建，本地表的存在是`Distributed`表创建的依赖条件，创建语句如下：

CREATE TABLE {teble} ON CLUSTER {cluster}
    AS {local_table}
    ENGINE= Distributed({cluster}, {
       database}, {local_table}，{policy})

这里的`policy`一般可以使用随机（例如`rand()`）或哈希（例如`halfMD5hash(id)`）

再来看下`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`表使用本地配置文件，在读取和写入时就不需要和`zookeeper`进行交互，并且`clickhouse`的集群配置是可以动态加载（2秒刷新一次）的，所以`Distributed`表在使用时很灵活，也可以为`Distributed`表建立多个`cluster`，因为`Distributed`表在查询时一般会向集群中所有的节点都发出请求，如果某个节点上没有相关联的本地表也会接到请求，如果可以将需要查询的节点配置一个集群可以避免一些空请求。当然`Distributed`表也可以通过`prewhere`或`where`设定条件来智能跳过不需要的`shard`。

### Distributed表写入原理 ###

有了上面的基础了解，就将进入主题了，本文主要是对`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`，表示写入是异步。

#### 同步写入还是异步写入 ####

同步写入是指数据直写入实际的表中，而异步写入是指数据首先被写入本地文件系统，然后发送到远端节点。

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<DistributedBlockOutputStream>(
            context, *this, createInsertToRemoteTableQuery(remote_database, remote_table, getSampleBlockNonMaterialized()), cluster,
            insert_sync, timeout);
    }

是否执行同步写入是由`insert_sync`决定的，最终是由是否配置`insert_distributed_sync`（默认为`false`）和`owned_cluster`值的或关系决定的，一般在使用`MergeTree`之类的普通表引擎时，通常是异步写入，但在使用表函数时（使用`owned_cluster`来判断是否是表函数），通常会使用同步写入。这也是在设计业务逻辑时需要注意的。

> #### `owned_cluster`是什么时候赋值的呢？ ####
> 
>     StoragePtr TableFunctionRemote::executeImpl(const ASTPtr & ast_function, const Context & context, const std::string & table_name) const
>     {
>         ......
>         StoragePtr res = remote_table_function_ptr
>                 ? StorageDistributed::createWithOwnCluster(
>                     table_name,
>                     structure_remote_table,
>                     remote_table_function_ptr,
>                     cluster,
>                     context)
>                 : StorageDistributed::createWithOwnCluster(
>                     table_name,
>                     structure_remote_table,
>                     remote_database,
>                     remote_table,
>                     cluster,
>                     context);
>         ......
>     }
>     StoragePtr StorageDistributed::createWithOwnCluster(
>         const std::string & table_name_,
>         const ColumnsDescription & columns_,
>         ASTPtr & remote_table_function_ptr_,
>         ClusterPtr & owned_cluster_,
>         const Context & context_)
>     {
>         auto res = create(String{}, table_name_, columns_, ConstraintsDescription{}, remote_table_function_ptr_, String{}, context_, ASTPtr(), String(), false);
>         res->owned_cluster = owned_cluster_;
>         return res;
>     }
> 
> 可以发现在创建`remote`表时会根据`remote_table_function_ptr`参数对最终的`owned_cluster_`赋值为`true`。

#### 异步写入是如何实现的 ####

了解了什么时候使用同步写入什么时候异步写入后，在继续分析正式的写入过程，同步写入一般场景中涉及较少，这里主要对异步写入逻辑进行分析。`outStream`的`write`方法如下：

void DistributedBlockOutputStream::write(const Block & block)
    {
        Block ordinary_block{ block };
        ......
        if (insert_sync)
            writeSync(ordinary_block);
        else
            writeAsync(ordinary_block);
    }

其实这个`write`方法是重写了`virtual void IBlockOutputStream::write(const Block & block)`，所以节点在接收到流并调用流的`write`方法就会进入该逻辑中。并且根据`insert_sync`来决定走同步写还是异步写。

#### 写入本地节点还是远端节点 ####

主要还是对异步写入进行分析，其实`writeAsync()`最终的实现方法是`writeAsyncImpl()`，如下：

void DistributedBlockOutputStream::writeAsyncImpl(const Block & block, const size_t shard_id)
    {
        const auto & shard_info = cluster->getShardsInfo()[shard_id];
    
        if (shard_info.hasInternalReplication())
        {
            if (shard_info.getLocalNodeCount() > 0)
            {
                /// Prefer insert into current instance directly
                writeToLocal(block, shard_info.getLocalNodeCount());
            }
            else
            {
                if (shard_info.dir_name_for_internal_replication.empty())
                    throw Exception("Directory name for async inserts is empty, table " + storage.getTableName(), ErrorCodes::LOGICAL_ERROR);
    
                writeToShard(block, {shard_info.dir_name_for_internal_replication});
            }
        }
        else
        {
            if (shard_info.getLocalNodeCount() > 0)
                writeToLocal(block, shard_info.getLocalNodeCount());
    
            std::vector<std::string> dir_names;
            for (const auto & address : cluster->getShardsAddresses()[shard_id])
                if (!address.is_local)
                    dir_names.push_back(address.toFullString());
    
            if (!dir_names.empty())
                writeToShard(block, dir_names);
        }
    }

其中`getShardsInfo()`方法就是获取`config.xml`配置文件中获取集群节点信息，`hasInternalReplication()`就对应着配置文件中的`internal_replication`参数，如果为`true`，就会进入最外层的if逻辑，否则就会进入else逻辑。其中`writeToLocal()`方法是相同的，是指如果`shard`包含本地节点，优先选择本地节点进行写入；后半部分`writeToShard()`就是根据`internal_replication`参数的取值来决定是写入其中一个远端节点，还是所有远端节点都写一次。

#### 数据如何写入本地节点 ####

当然一般情况`Distributed`表还是基于`ReplicatedMergeTree`系列表进行创建，而不是基于表函数的，所以大多数场景还是会先写入本地再分发到远端节点。那写入`Distributed`表的数据是如何保证原子性落盘而不会在数据正在写入的过程中就把不完整的数据发送给远端其他节点呢？看下`writeToShard()`方法，如下：

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);
        }
    
        /** remove the temporary file, enabling the OS to reclaim inode after all threads
            *  have removed their corresponding files */
        Poco::File(first_file_tmp_path).remove();
    }

首先来了解下`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`命名，是为了在后续分发到远端节点保持顺序性。

#### 数据如何分发到各个节点 ####

细心的已经发现在`writeToShard()`方法中有个`requireDirectoryMonitor()`，这个方法就是将`shard`目录注册监听，并通过专用类`StorageDistributedDirectoryMonitor`来实现数据文件的分发，根据不同配置可以实现逐一分发或批量分发。并且包含对坏文件的容错处理。

--------------------

#### 重点来啦！ ####

分析到这，可能还有人会觉得云里雾里，觉得整个流程串不起来，其实这样写是为了先不影响`Distributed`表写入的主流程，明白了这个再附加上`sharding_key`拆分和权重拆分就很好理解了。

#### 如何根据`sharding_key`和`weight`拆分数据 ####

上面提到过`writeAsync()`的最终实现方法是`writeAsyncImpl`，这个说法是没问题的，但是中间还有段关键逻辑，如下：

void DistributedBlockOutputStream::writeAsync(const Block & block)
    {
        if (storage.getShardingKeyExpr() && (cluster->getShardsInfo().size() > 1))
            return writeSplitAsync(block);
    
        writeAsyncImpl(block);
        ++inserted_blocks;
    }
    
    void DistributedBlockOutputStream::writeSplitAsync(const Block & block)
    {
        Blocks splitted_blocks = splitBlock(block);
        const size_t num_shards = splitted_blocks.size();
    
        for (size_t shard_idx = 0; shard_idx < num_shards; ++shard_idx)
            if (splitted_blocks[shard_idx].rows())
                writeAsyncImpl(splitted_blocks[shard_idx], shard_idx);
    
        ++inserted_blocks;
    }

`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()`方法实现，如下：

Blocks DistributedBlockOutputStream::splitBlock(const Block & block)
    {
        auto selector = createSelector(block);
        ......
        for (size_t col_idx_in_block = 0; col_idx_in_block < columns_in_block; ++col_idx_in_block)
        {
            MutableColumns splitted_columns = block.getByPosition(col_idx_in_block).column->scatter(num_shards, selector);
            for (size_t shard_idx = 0; shard_idx < num_shards; ++shard_idx)
                splitted_blocks[shard_idx].getByPosition(col_idx_in_block).column = std::move(splitted_columns[shard_idx]);
        }
    
        return splitted_blocks;
    }
    
    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<UInt8>()`）。

到此，对于`Distributed`表的写入流程的关键点就大致分析完了。篇幅有限有些细节没有做过多说明，有兴趣的可以自行再了解下。

### 总结 ###

通过对`Distributed`表写入流程的分析，了解了该类型表的实际工作原理，所以在实际应用中有几个点还需要关注一下：

1.  `Distributed`表在写入时会在本地节点生成临时数据，会产生写放大，所以会对CPU及内存造成一些额外消耗，建议尽量少使用`Distributed`表进行写操作。
2.  `Distributed`表写的临时block会把原始block根据sharding\_key和weight进行再次拆分，会产生更多的block分发到远端节点，也增加了merge的负担。
3.  `Distributed`表如果是基于表函数创建的，一般是同步写，需要注意。

了解原理才能更好的使用，遇到问题才能更好的优化。