Citus总结:架构介绍

2022-04-13 00:00:00 数据 分布式 节点 分片 副本

1. 整体架构

Citus的架构如上图所示,为Share Nothing的分布式架构,包含两种类型的节点:

  • Coordinator协调者节点

    • 负责接收来自客户端的交互请求,并把请求按照规则转发给对应Worker节点执行;

    • 本地维护Metadata元数据,例如Shard分片信息,放置节点信息等;

  • Worker工作节点,也叫数据节点

    • 负责用户数据表的实际存储

Citus通过Extension的方式扩展了PostgreSQL下面的功能:

  • System catalogs:新增分布式元数据表等

  • Planner hook:新增"增删改查"操作的分布式查询计划

  • Executor hook:新增"增删改查"操作的分布式执行逻辑

  • Utility hook:修改Alter Table,Create Index,Vacuum等逻辑支持分布式;

  • Transaction & resource handling:支持同时打开更多文件描述符

  • Background worker process:新增后台任务进程如:死锁检测,任务跟踪进程等;

  • Logical decoding:支持在线数据迁移

2. 支持的表类型

2.1 Distributed Tables 分布式表

分布式表是Citus中常见的表,一般建议把大于10GB的业务表创建为分布式表,以支持更大的数据量,以及可扩展的读写性能。

分布式表的数据按照指定分布式列的Hash值打散到多个Worker节点,每个Shard分片都有的shardid编号,每个Shard对应Worker节点上名为tablename_shardid
的普通表,该表并与Coordiantor节点上的表具有相同的表模式、索引和约束定义。

在创建分布式表时,选择合适的分布式列非常重要,一般是选择如tenant_id
字段将不同租户的数据打散存储在不同节点,下面两种情况的列不建议选择为分布式列:

  • Distinct结果非常多列,如时序数据中的timestamp字段,不适合用作分布式列,会导致关联数据没有亲和性;当然太少的列也不行,如果少于Worker节点数则会导致节点负载不均衡;

  • 取值分布严重倾斜的列,会导致Worker节点负载不均衡;

分布式表的创建方式如下:

    CREATE TABLE github_events
    (
    event_id bigint,
    event_type text,
    event_public boolean,
    repo_id bigint,
    payload jsonb,
    repo jsonb,
    actor jsonb,
    org jsonb,
    created_at timestamp
    );


    SELECT create_distributed_table('github_events', 'repo_id');

    创建分布式表不仅可以用于空表,也可用于已经存储有数据的表,Citus会自动按照Sharding规则将表打散到多个Worker节点。

    2.1.1 数据Sharding分片

    具体步骤如下:

    1. 针对每行数据,计算分布式列值的Hash值;

    2. 对Hash值进行Range切割,每一个Range对应一个Shard,切割的边界值保存在Coordinator节点的pg_dist_shard
      表中;

    3. 再将Shard映射到不同的Worker/Data节点,映射关系保存在Coordinator节点的pg_dist_placement
      表中;

    元数据表pg_dist_shard
    的内容如下:

      SELECT * from pg_dist_shard;


      logicalrelid | shardid | shardstorage | shardminvalue | shardmaxvalue
      ---------------+---------+--------------+---------------+---------------
      github_events | 102026 | t | 268435456 | 402653183
      github_events | 102027 | t | 402653184 | 536870911
      github_events | 102028 | t | 536870912 | 671088639
      github_events | 102029 | t | 671088640 | 805306367
      (4 rows)

      其中logicalrelid
      为用户的分布式表名,shardminvalue
      shardmaxvalue
      为Hash值的Range范围,落入这范围的行归宿对应的shardid

      通过联合查询pg_dist_placement
      pg_dist_node
      可以查看shard对应的节点信息:

        SELECT
        shardid,
        node.nodename,
        node.nodeport
        FROM pg_dist_placement placement
        JOIN pg_dist_node node
        ON placement.groupid = node.groupid
        AND node.noderole = 'primary'::noderole
        WHERE shardid = 102027;


        ┌─────────┬───────────┬──────────┐
        │ shardid │ nodename │ nodeport │
        ├─────────┼───────────┼──────────┤
        │ 102027 │ localhost │ 5433 │
        └─────────┴───────────┴──────────┘

        通过元数据映射的方式,将不同shard放置到不同的worker节点。

        2.1.2 分片数量和副本数

        分布式表有两个重要的配置参数:

        • 分片数量 citus.shard_count

          • 指定分布式表要被切分成多少个Shard,默认配置为32个Shard;

          • 一般是建议配置为CPU总核数的2~4倍;

        • 副本数 citus.shard_replication_factor

          • Citus还支持多副本存储Shard,并提供副本数量的配置参数;

          • 在多副本存储情况下,同一个shard的不同副本使用相同的shardid,pg_dist_placement
            表中针对每个副本都有一条记录;

          • 这里不同于PostgreSQL的Stream Replication,是Citus提供的副本能力,当然也可以选择使用Stream Replication而不是Citus的副本功能,将此处的副本数配置为1;

        2.1.3 Co-Location 亲和性共存

        如果不同分布式表所指定的分布式列类型、shard分区数量及副本数量都相同,则这些分布式表都会按照相同的Hash值范围分片,并把相同范围的分片存储在相同的Worker节点上,这样分布式列值相同的行会亲和性的存储在同一个节点。

        例如,通过下面方式创建的分布式表存在亲和性关系:

          -- these tables are implicitly co-located by using the same
          -- distribution column type and shard count with the default
          -- co-location group


          SELECT create_distributed_table('events', 'some_int_col');
          SELECT create_distributed_table('page', 'other_int_col');

          events
          page
          两个表在Worker节点的布局示例如图:

          创建分布式表时,也可以colocate_with
          参数主动指定和某个表亲和性共存:

            -- distribute stores
            SELECT create_distributed_table('stores', 'store_id');


            -- add to the same group as stores
            SELECT create_distributed_table('orders', 'store_id', colocate_with => 'stores');
            SELECT create_distributed_table('products', 'store_id', colocate_with => 'stores');

            不指定colocate_with
            参数时默认为default
            ,和所有具有相同分布式列类型、shard分区数量及副本数量的分布式表亲和性共存。

            Coordinator节点通过元数据表pg_dist_colocation
            维护亲和性共存关系:

              SELECT * from pg_dist_colocation;
              colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
              --------------+------------+-------------------+------------------------+-----------------------------
              2 | 32 | 2 | 20 | 0
              (1 row)

              其中colocationid
              为co-location组id,其他为该组的属性字段。

              通过将同一个租户的不同业务表进行亲和性共存,能带来以下好处:

              • 同一租户的相关数据都在同一个节点,查询单个租户时,可以支持完整的SQL能力;

              • 支持在同一个包含多条语句的事务中修改co-located相关shards;

              • 支持INSERT..SELECT语句

              • co-located shards间支持Foreign keys关系;

              • 支持分布式outer joins

              • Pushdown CTEs

              一旦分布式表形成亲和性共存关系后,后续在集群扩容数据重均衡时,Citus也会保证亲和性共存关系,会相关shard一起迁移到新节点。

              2.1.4 Append分片方式

              除了Hash分片方式外,分布式表还支持Append分片方式,只适用于append only的负载如时序数据场景,新增数据都追加写入到新一个shard分片,新增分片以轮询的方式创建在不同的Worker节点上,因此实时只有一个Worker节点在写入。

                SELECT create_distributed_table('github_events', 'created_at', 'append');

                实际上,即使是时序数据场景也很少用Append分片方式,一般都结合PostgreSQL分区表 + Citus Hash分片
                的方式适用,先按照时间列进行范围分区,再指定设备Id为分片列进行Hash打散。

                2.2 Reference Tables 引用表

                引用表不会进行Sharding分片,所有数据都在一个Shard内,并在所有Worker节点都复制存储一份,Citus通过2PC保证多个副本的强一致性。

                引用表一般用于和具体租户无关的字典类的小表,这类表需要经常和其他业务表进行Join查询,因为所有Worker节点都可以本地访问到引用表数据,可以避免在Join查询时跨节点去获取数据。

                引用表的创建方式如下:

                  -- a reference table
                  CREATE TABLE states (
                  code char(2) PRIMARY KEY,
                  full_name text NOT NULL,
                  general_sales_tax numeric(4,3)
                  );


                  -- distribute it to all workers
                  SELECT create_reference_table('states');

                  创建引用表时,不需要指定分布式列。

                  2.3 Local Tables 本地表

                  指存储在Coordinator节点本地的表,一般仅限于不需要和其他表进行Join查询的表,如用户身份认证的表,Citus自身的元数据表也可以认为是本地表。

                  3. 参考

                  http://docs.citusdata.com/en/v10.0/get_started/concepts.html

                  http://docs.citusdata.com/en/v10.0/develop/reference_ddl.html

                  http://docs.citusdata.com/en/v10.0/sharding/data_modeling.html


                  相关文章