Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Maghead: Horizontal Scaling Database by Sharding

Maghead: Horizontal Scaling Database by Sharding

Maghead - A high performance database framework that supports Application level sharding

Yo-An Lin

July 01, 2017
Tweet

More Decks by Yo-An Lin

Other Decks in Technology

Transcript

  1. Outline • Maghead Overview • Why? And the history •

    The look of Maghead • Performance • Sharding • When do you Shard? • What's Sharding? • How to Shard? • Consistent Hashing • Chunks and Chunk Migration • Shard Balancing • Commands and the APIs
  2. Actually if you • need to finish 2+ site projects

    per month • have different modules shared between different projects with some customization • need to maintain legacy projects without db query tests • have frequently schema changes from different team members • need to remember more than the changes from more than 10, 20, 30 tables
  3. ORM helps you ... • Have the domain logics well-organized.

    • Perform lightweight automatic migration when you have schema changes. • Generate the common, basic SQL queries for the most use cases without the need of manually updating the SQL queries. • Manage the database connections with a centralized simple config file. • !! Actually you can still do hand-written query if you want
  4. --- cli: bootstrap: vendor/autoload.php schema: auto_id: true finders: - {

    name: ComposerSchemaFinder, args: ["composer.json"] }
  5. --- ... instance: local: dsn: 'mysql:host=localhost' user: root driver: mysql

    host: localhost password: null query_options: { } connection_options: 1002: 'SET NAMES utf8' ...
  6. --- ... databases: master: dsn: 'mysql:host=localhost;dbname=testing' host: localhost user: root

    driver: mysql database: testing password: null query_options: { } connection_options: 1002: 'SET NAMES utf8'
  7. Bootstrap Code require 'vendor/autoload.php'; use Maghead\Runtime\Config\FileConfigLoader; use Maghead\Runtime\Bootstrap; $config =

    FileConfigLoader::load( __DIR__ . '/db/config/ database.yml'); Bootstrap::setup($config);
  8. Bootstrap Code require 'vendor/autoload.php'; use Maghead\Runtime\Config\FileConfigLoader; use Maghead\Runtime\Bootstrap; $config =

    FileConfigLoader::load('db/config/database.yml'); Bootstrap::setup($config); 2 lines only!
  9. namespace Todos\Model; use Maghead\Schema\DeclareSchema; class TodoSchema extends DeclareSchema { public

    function schema() { $this->column('title') ->varchar(128) ->required() ; $this->column('done') ->boolean() ->default(false); $this->column('description') ->text(); $this->column('created_at') ->timestamp() ->default(function() { return date('c'); }); } }
  10. namespace Todos\Model; class TodoSchemaProxy extends RuntimeSchema { const SCHEMA_CLASS =

    'Todos\\Model\\TodoSchema'; const LABEL = 'Todo'; const MODEL_NAME = 'Todo'; const MODEL_NAMESPACE = 'Todos\\Model'; const MODEL_CLASS = 'Todos\\Model\\Todo'; const REPO_CLASS = 'Todos\\Model\\TodoRepoBase'; const COLLECTION_CLASS = 'Todos\\Model\\TodoCollection'; const TABLE = 'todos'; const PRIMARY_KEY = 'id'; const GLOBAL_PRIMARY_KEY = NULL; const LOCAL_PRIMARY_KEY = 'id'; public static $column_hash = array ( 'id' => 1, 'title' => 1, 'done' => 1, 'description' => 1, 'created_at' => 1, ); public $columnNames = array ( 0 => 'id', 1 => 'title', 2 => 'done', 3 => 'description', 4 => 'created_at', ); ............ cut
  11. class ProductRepoBase extends Repo { const FIND_BY_PRIMARY_KEY_SQL = 'SELECT *

    FROM products WHERE id = ? LIMIT 1'; const DELETE_BY_PRIMARY_KEY_SQL = 'DELETE FROM products WHERE id = ?'; const FETCH_CREATED_BY_SQL = 'SELECT * FROM users WHERE id = ? LIMIT 1'; const FETCH_UPDATED_BY_SQL = 'SELECT * FROM users WHERE id = ? LIMIT 1'; const FETCH_PRODUCT_FEATURES_SQL = 'SELECT * FROM product_feature_junction WHERE product_id = ?'; const FETCH_PRODUCT_PRODUCTS_SQL = 'SELECT * FROM product_products WHERE product_id = ?'; const FETCH_IMAGES_SQL = 'SELECT * FROM product_images WHERE product_id = ?'; const FETCH_PROPERTIES_SQL = 'SELECT * FROM product_properties WHERE product_id = ?'; const FETCH_TYPES_SQL = 'SELECT * FROM product_types WHERE product_id = ?'; const FETCH_RESOURCES_SQL = 'SELECT * FROM product_resources WHERE product_id = ?'; const FETCH_FILES_SQL = 'SELECT * FROM product_files WHERE product_id = ?'; const FETCH_PRODUCT_RECIPES_SQL = 'SELECT * FROM recipe_to_products WHERE product_id = ?'; const FETCH_CATEGORY_SQL = 'SELECT * FROM product_categories WHERE id = ? LIMIT 1'; const FETCH_PRODUCT_TAGS_SQL = 'SELECT * FROM product_tag_junction WHERE product_id = ?';
  12. class ProductRepoBase extends Repo { public function deleteByPrimaryKey($pkId) { if

    (!$this->deleteStm) { $this->deleteStm = $this->write->prepare(self::DELETE_BY_PRIMARY_KEY_SQL); } return $this->deleteStm->execute([$pkId]); } public function fetchCreatedByOf(Model $record) { if (!$this->fetchCreatedByStm) { $this->fetchCreatedByStm = $this->read->prepare(self::FETCH_CREATED_BY_SQL); $this->fetchCreatedByStm->setFetchMode(PDO::FETCH_CLASS, \UserBundle\Model\User::class, [$this]); } $this->fetchCreatedByStm->execute([$record->created_by]); $obj = $this->fetchCreatedByStm->fetch(); $this->fetchCreatedByStm->closeCursor(); return $obj; }
  13. $create = $product->asCreateAction(); {{ RecordAction.renderSignatureWidget|raw }} {{ RecordAction.renderCSRFTokenWidget|raw }} {%

    if Record.hasKey %} {{RecordAction.renderKeyWidget|raw}} {% endif %} {{RecordAction.renderField('account') |raw }} {{RecordAction.renderField('password') |raw }}
  14. $record = new User; $record->account = "c9s"; $record->password = sha1("mypassword");

    $record->save(); Different States:
 1. existing record 2. non-existing record (before inserting the row)
  15. function updatePassword($user) { // check the existence because the object

    // might be not able to update the // password if (!$user->hasKey()) { throw new LogicException; } // update ... }
  16. function updatePassword($user) { // check the existence because the object

    // might be not able to update the // password if (!$user->hasKey()) { throw new LogicException; } // update ... } The user record might be not created yet.
  17. function updatePassword($user) { // check the existence because the object

    // might be not able to update the // password if (!$user->hasKey()) { throw new LogicException; } // update ... } Check the key to verify the existence
  18. $record = User::load([ "account" => "timcook" ]); If the returned

    $record is not false, then it must be an existing row.
  19. $user = User::createAndLoad([ "account" => "timcook" ]); If the returned

    $record is not false, then it must be an existing row.
  20. use Doctrine\DBAL\DriverManager; $conn = DriverManager::getConnection(array( 'wrapperClass' => 'Doctrine\DBAL\Connections\MasterSlaveConnection', 'driver' =>

    'pdo_mysql', 'keepSlave' => true, 'master' => array( 'user' => 'ideato', 'password' => 'ideato', 'dbname' => 'db_ideato' ), 'slaves' => array( array( 'user' => 'ideato', 'password' => 'ideato', 'dbname' => 'db_ideato_slave' ) ) ) ); $entityManager = EntityManager::create($conn, $config);
  21. // https://gist.github.com/ricfrank/d6f6317a1a1434cdc364 $entityManager = EntityManager::create($conn, $config); $productRepository = $entityManager->getRepository('Product'); $masterSlaveConn->connect('slave');

    $productRepository = $entityManager->getRepository('Product'); $masterSlaveConn->connect('master'); $productRepository = $entityManager->getRepository('Product');
  22. $entityManager = EntityManager::create($conn, $config); $productRepository = $entityManager->getRepository('Product'); $masterSlaveConn->connect('slave'); $product =

    new Product(); $product->setName("nuovo_prod1"); $product->setCategory("oeeooeoe"); $entityManager->persist($product); $entityManager->flush(); $productRepository = $entityManager->getRepository('Product'); $masterSlaveConn->connect('master'); $productRepository = $entityManager->getRepository('Product');
  23. $ret = Product::repo("master") ->create([ "name" => "Samsung S3" ]); $ret

    = Product::masterRepo() ->create([ "name" => "Samsung S3" ]);
  24. 0 125 250 375 500 PDO Maghead LessQL YiiM Yii2M

    Propel20 DoctrineM Eloquent Benchmark: score -> lower is better
  25. 0 125 250 375 500 PDO Maghead LessQL YiiM Yii2M

    Propel20 DoctrineM Eloquent Insertion
  26. 0 125 250 375 500 PDO Maghead LessQL YiiM Yii2M

    Propel20 DoctrineM Eloquent findPK
  27. 0 125 250 375 500 PDO Maghead LessQL YiiM Yii2M

    Propel20 DoctrineM Eloquent Complex Query
  28. 0 100 200 300 400 PDO Maghead LessQL YiiM Yii2M

    Propel20 DoctrineM Eloquent Benchmark: Insert lower is better
  29. 0 15 30 45 60 PDO Maghead LessQL YiiM Yii2M

    Propel20 DoctrineM Eloquent Benchmark: Find PK lower is better
  30. 0 125 250 375 500 PDO Maghead LessQL YiiM Yii2M

    Propel20 DoctrineM Eloquent Benchmark: Complex Query lower is better
  31. 100,000 records 100,000 records 100,000 records 100,000 records 100,000 records

    100,000 records 100,000 records 100,000 records 100,000 records 100,000 records 10,000,000 records
  32. app_db_3 Server #1 app_db_2 app_db_1 app_db_0 Shard #1 (user id

    0~100) Shard #2 (user id 101~200) Shard #3 (user id 201~300) Shard #4 (user id 301~400)
  33. app_db_3 Server #1 app_db_2 app_db_1 app_db_0 Shard #1 Shard #2

    Shard #3 Shard #4 app_db_7 Server #2 app_db_6 app_db_5 app_db_4 Shard #5 Shard #6 Shard #7 Shard #8 Multiple Shards per Server
  34. Server #1 app_db_0 Shard #1 Server #2 app_db_1 Shard #2

    Server #3 app_db_2 Shard #3 Server #4 app_db_3 Shard #4 One Shard per Server
  35. Key

  36. Example #1 • Company (id) • Store (id, company_id) •

    Order (id, store_id, amount ...) • OrderItem (order_id, quantity, ...) • OrderPayment (order_id...)
  37. Example #1 • Company (id) • Store (id, company_id) •

    Order (id, company_id, store_id, amount ...) • OrderItem (company_id, store_id, order_id, quantity, ...) • OrderPayment (company_id, store_id, order_id ...)
  38. Different Sharding Approaches • Database level sharding • SQL level

    sharding • Middleware-based sharding • Application level sharding
  39. RDBMS Solutions • Vitas (from YouTube, Written in Go) •

    MySQL Cluster (NDB) • MySQL Fabric (Middleware-based sharding in Python) • MariaDB Spider Engine • GPDB (PostgreSQL)
  40. Vitas • Written in Go • Use sql parser to

    distribute the queries • Only for MySQL • protocol gRPC • Used by YouTube
  41. MySQL Cluster • Limited to 48 hosts • Shard Key

    MUST BE in primary key and can't be changed. Similar to MySQL partitions. • Primary key could be compound primary key • NDB Engine • When query doesn't include shard key, query will be scattered query.
  42. MySQL Fabric • Middleware-based sharding • Written in Python •

    Has it's own connector API binding for different languages (Java, Python) • Manually set the host to operate from the application code. • Bottleneck will be the middleware.
  43. foreach ($mapping->chunks as $i => $c) { $x = $c['index'];

    $this->buckets[$x] = $i; $this->targetIndexes[$i][] = $x; } ksort($this->buckets, SORT_REGULAR);
  44. • Key(0~100) Chunk #1 • Key(101~200) Chunk #2 • Key(201~300)

    Chunk #3 • Key(301~400) Chunk #4 • ........
  45. • Key(0~100) Chunk #1 -> Shard #1 • Key(101~200) Chunk

    #2 -> Shard #1 • Key(201~300) Chunk #3 -> Shard #2 • Key(301~400) Chunk #4 -> Shard #2 • ........
  46. $mapping = new ShardMapping('store_keyspace', [ 'key' => 'store_id', 'shards' =>

    ['node1', 'node2', 'node3'], 'chunks' => [ ["from" => 0, "index" => 536870912, "shard" => "node1" ], ["from" => 536870912, "index" => 1073741824, "shard" => "node1" ], ["from" => 1073741824, "index" => 1610612736, "shard" => "node1" ], ["from" => 1610612736, "index" => 2147483648, "shard" => "node2" ], ["from" => 2147483648, "index" => 2684354560, "shard" => "node2" ], ["from" => 2684354560, "index" => 3221225472, "shard" => "node2" ], ["from" => 3221225472, "index" => 3758096384, "shard" => "node3" ], ["from" => 3758096384, "index" => 4294967296, "shard" => "node3" ], ] ], $dataSourceManager); $hasher = new FastHasher($mapping); $hash = $hasher->hash($key); // return 3221225472
  47. • Server #1 (10 shards) • Server #2 (10 shards)

    • Server #3 (10 shards) Server load increases because server #1 has more rows than other servers.
  48. • Server #1 (9 shards) • Server #2 (10 shards)

    • Server #3 (10 shards) • Server #4 (1 shard) Migrate the bottleneck shard from Server #1 to a new server.
  49. use Maghead\Sharding\Operations\CloneShard; $op = new CloneShard($config); // public function clone($mappingId,

    $instanceId, $newNodeId, $srcNodeId) $op->clone("store_keyspace", "server3", "shard1000", "shard0003");
  50. maghead shard clone --drop-first --mapping store_key --instance server2 a01 a11

    clone shard "a01" to the a new shard on server2 named "a11"
  51. maghead shard allocate --mapping store_key --instance server2 a02 allocate an

    empty shard "a02" on server 2 and initialize all the tables
  52. $balancer = new ShardBalancer( new ConservativeShardBalancerPolicy(1, false, 1.3) ); $migrateInfo

    = $balancer->balance($mapping, [new OrderSchema]); 保守式平衡規則
  53. $balancer = new ShardBalancer( new ConservativeShardBalancerPolicy(1, false, 1.3) ); $migrateInfo

    = $balancer->balance($mapping, [new OrderSchema]); ⼀一次只找出⼀一個 Chunk 做 Migration
  54. $balancer = new ShardBalancer( new ConservativeShardBalancerPolicy(1, false, 1.3) ); $migrateInfo

    = $balancer->balance($mapping, [new OrderSchema]); 找出 rows 總量量超過平均 1.3 倍的 Shard
  55. • Key(0~100) Chunk #1 -> Shard #1 • Key(101~200) Chunk

    #2 -> Shard #1 • Key(201~300) Chunk #3 -> Shard #2 • Key(301~400) Chunk #4 -> Shard #2 • ........
  56. • Key(0~100) Chunk #1 -> Shard #1 • Key(101~200) Chunk

    #2 -> Shard #1 • Key(201~300) Chunk #3 -> Shard #2 • Key(301~400) Chunk #4 -> Shard #2 • ........ When the load of shard #2 increases, we can assign some chunks to the other shard.
  57. • Key(0~100) Chunk #1 -> Shard #1 • Key(101~200) Chunk

    #2 -> Shard #1 • Key(201~300) Chunk #3 -> Shard #2 • Key(301~400) Chunk #4 -> Shard #2 • ........ When the rows in Chunk #4 grow up too much, We can split the Chunk #4 into 2+ chunks and assign them to the other shards.
  58. Chunk #1 Chunk #2 Chunk #3 Chunk #4 Chunk #5

    Chunk #6 Chunk #7 Chunk #8 Shard #1 Shard #2 Chunk #4 Copy
  59. Chunk #1 Chunk #2 Chunk #3 Chunk #4 Chunk #5

    Chunk #6 Chunk #7 Chunk #8 Shard #1 Shard #2 Chunk #4 Verify
  60. Chunk #1 Chunk #2 Chunk #3 Chunk #5 Chunk #6

    Chunk #7 Chunk #8 Shard #1 Shard #2 Chunk #4 Remove
  61. namespace Maghead\Sharding\Manager; class ChunkManager { public function __construct(ShardMapping $mapping) {

    } public function distribute(array $shardIds, $numberOfChunks = 32) { } public function verify(Chunk $chunk, Shard $dstShard, array $schemas) { } public function remove(Chunk $chunk, array $schemas) { } public function clone(Chunk $chunk, Shard $dstShard, array $schemas) { } public function migrate(Chunk $chunk, Shard $dstShard, array $schemas) { } public function move(Chunk $chunk, Shard $dstShard, array $schemas) { } public function split(Chunk $chunk, $n = 2) { } }
  62. --- sharding: mappings: M_store_id: key: "store_id" shards: - node1 -

    node2 - node3 hash: true chunks: - { from: 0, index: 536870912, shard: node1 } - { from: 536870912, index: 1073741824, shard: node1 } - { from: 1073741824, index: 1610612736, shard: node1 } - { from: 1610612736, index: 2147483648, shard: node2 } - { from: 2147483648, index: 2684354560, shard: node2 } - { from: 2684354560, index: 3221225472, shard: node2 } - { from: 3221225472, index: 3758096384, shard: node3 } - { from: 3758096384, index: 4294967296, shard: node3 } ⾃自動分派的 Chunk 定義
  63. class StoreSchema extends DeclareSchema { public function schema() { $this->globalTable("M_store_id");

    Store table 需要在 Shards 中被 Join 所以被定義為 Global Table
  64. namespace StoreApp\Model; use Maghead\Schema\DeclareSchema; class OrderSchema extends DeclareSchema { public

    function schema() { $this->column('uuid', 'uuid-pk'); {....cut....} $this->shardBy("M_store_id"); } } Order table 需要依照定義的 keyspace 做分片
  65. namespace StoreApp\Model; use Maghead\Schema\DeclareSchema; class OrderSchema extends DeclareSchema { public

    function schema() { $this->column('uuid', 'uuid-pk'); {....cut....} $this->shardBy("M_store_id"); } } 快速定義 UUID primary key
  66. $ret = Order::create([ 'store_id' => $storeId, 'amount' => 600, ]);

    Automatically dispatch record to the shard by the store_id
  67. $query = new SelectQuery; $query->select(['SUM(amount)' => 'amount']); $query->from('orders'); $query->where()->in('store_id', [1,2]);

    $client = new GearmanClient; $client->addServer(); $mapper = new GearmanQueryMapper($client); $res = $mapper->map($shards, $query);
  68. $rows = [ [ 'category' => 'Food', 'type' => 'pasta',

    'amount' => 1 ], [ 'category' => 'Food', 'type' => 'pasta', 'amount' => 1 ], [ 'category' => 'Food', 'type' => 'juice', 'amount' => 1 ], [ 'category' => 'Food', 'type' => 'juice', 'amount' => 1 ], [ 'category' => 'Book', 'type' => 'programming', 'amount' => 5 ], [ 'category' => 'Book', 'type' => 'programming', 'amount' => 2 ], [ 'category' => 'Book', 'type' => 'cooking', 'amount' => 6 ], [ 'category' => 'Book', 'type' => 'cooking', 'amount' => 2 ], ]; $result = group_by($rows, ['category','type'], [ 'total_amount' => [ 'selector' => 'amount', 'aggregator' => REDUCER_AGGR_SUM, ], 'cnt' => REDUCER_AGGR_COUNT, ]); print_r($result);