AmebaブログにおけるDynamoDBからMySQLへのストレージ移行の取り組み

こんにちは、26卒内定者でバックエンドエンジニアの相馬大和です！ 本記事では、内定者バイト期間中の、Amebaブログのストレージ移行と、それに伴う周辺システム刷新の取り組みをご紹介します。 Amebaブログは様々なコンポーネントから構成されており、その多くでDynamoDBを利用しています。 DynamoDBはスケーラビリティに優れたNoSQLデータベースであり、かつてのAmebaブログの要件には適していました。しかし、サービスの成長とともにデータの参照パターンが多様化した結果、現在の運用においてはRDBの方が適しているという判断に至りました。 その背景として、まず運用負荷の観点があります。DynamoDBのスループット制御はWCU（Write Capacity Unit）によって管理されますが、障害復旧時のリトライ処理においてWCUの調整が難しく、運用負荷となっていました。 次に、データの参照パターンの多様化への対応です。ブログのデータは多方面で活用されることが多く、アクセスパターンが増えるたびにGSIの追加を検討する必要がありました。RDBであれば、こうした柔軟なクエリへの対応がより低コストで実現できます。 三つ目はコスト面です。現在のアクセス規模においては、Aurora MySQLの課金モデルの方がより費用対効果が高く、既存の月額約285ドルから約2ドルへの大幅なコスト削減が見込まれます。 これらを踏まえ、Amebaブログ全体としてDynamoDBからRDBへの移行が計画されており、その第一歩目として、ハッシュタグ記事ランキングを管理するストレージを移行しました。 Amebaブログでは記事ごとにハッシュタグをつけることができ、ハッシュタグのランキングを閲覧することが可能です。 ![](https://developers.cyberagent.co.jp/blog/wp-content/uploads/2026/02/d41956aa4b4d616567338376a4030f63-1-300x262.jpg) Amebaブログのハッシュタグ記事ページ https://blogtag.ameba.jp/ 今回の内定者バイトでは、このハッシュタグ記事のランキングデータを管理するDynamoDBをMySQLに置き換えると同時に、DynamoDBに依存するリード系APIや、データを書き込むバッチ処理も再実装しました。 まず既存実装の流れをご紹介します。 ![](https://developers.cyberagent.co.jp/blog/wp-content/uploads/2026/02/b3bda72a146237ba876418441c1908f2-300x110.png) DSC (Data Science Center)の方が管理してくださっている、BigQueryの元データを、Amebaブログ側のEKSのCronJobがstreamし、DynamoDBに書き込んでいます。そのDynamoDBを参照する、Goで書かれたAPIがElasticBeanstalkで動いており、ハッシュタグ記事ランキング関連のデータを返しています。 そして今回の新実装では、以下のような流れになります。 ![](https://developers.cyberagent.co.jp/blog/wp-content/uploads/2026/02/449fd1ad0c7e527b90925f811d46871f-300x102.png) 上記構成のように、ハッシュタグ記事ランキングのデータを管理するDynamoDBをMySQLに移行するためには、以下三つのタスクを行いました。 （ElasticBeanstalk → EC2の移行は後ほど解説します。） DynamoDBのデータ構造をもとに、RDB用のテーブルを設計する ハッシュタグ記事ランキングのデータをBigQueryからstreamし、MySQLに書き込むバッチを再実装し、EKSのCronjobで動かす 旧DynamoDBのテーブルを参照していたAPIをElasticBeanstalkからEC2に移行する まず最初に行ったのが、DynamoDBのデータ構造をもとに、RDB用のテーブルを設計し、構築することです。 DynamoDBでは、キーの検索効率を上げるために複数の属性を連結した複合文字列（例：ameba_entry_id_tag_name MySQL移行にあたって、これらの複合キーを分解し、リレーショナルモデルとして適切に正規化した設計に刷新しました。 次にインデックスの再定義に関してです。GSIをもとに、RDBのインデックス設計を行いましたが、この対応でコスト面でメリットを享受できました。理由としては、DynamoDBのGSIは、メインテーブルへの書き込みのたびにGSI分の書き込みも独立して発生するため、インデックス数に比例してWCUを余分に消費する一方で、MySQLのセカンダリインデックスは、インデックス数に応じてコストが線形増加するような課金構造ではないためです。 最後にDB、テーブル構築に関しては、今回すでに利用されているRDSのクラスターの中に作成し、Batchでの書き込みはプライマリインスタンス、DBから読み込む際は、リードレプリカを参照するようにしました。 ハッシュタグ記事ランキングの元データはBigQueryに蓄積されており、ここから毎日1.5GBほどのデータを抽出し、MySQLへ反映させるバッチ処理の再実装を行いました。 BQからのstream処理では、1.5GBのデータを一度にメモリに載せると、アウトオブメモリでkillされてしまうため、Goの並列処理（ChannelとGoroutine）を利用して、「BigQueryからの読み出し」と「MySQLへの書き込み」をオーバーラップさせるパイプラインを構築しました。 func StreamHashtagEntryRankingFromBQ(ctx context.Context, bqClient *bigquery.Client) (<-chan *HashtagEntryRanking, <-chan error, error) { err := WaitForTableActive(ctx, bqClient, "prj_hashtag_entry_ranking", "hashtag_ranking_similarity_normalized_pv") if err != nil { return nil, nil, fmt.Errorf("WaitForTableActive failed: %w", err) } q := bqClient.Query(HashtagEntryRankingListQuery) job, err := q.Run(ctx) if err != nil { return nil, nil, fmt.Errorf("job run failed: %w", err) } status, err := job.Wait(ctx) if err != nil { return nil, nil, fmt.Errorf("job wait failed: %w", err) } if status.Err() != nil { return nil, nil, fmt.Errorf("job execution error: %w", status.Err()) } it, err := job.Read(ctx) if err != nil { return nil, nil, fmt.Errorf("job read failed: %w", err) } // 1ページあたり500行ずつ取得する it.PageInfo().MaxSize = 500 // データを逐次的に返すチャネルを作成 outChan := make(chan *HashtagEntryRanking) errChan := make(chan error, 1) go func() { defer close(outChan) defer close(errChan) for { bqRanking := new(HashtagEntryRanking) err := it.Next(bqRanking) if errors.Is(err, iterator.Done) { logger.Default().Info("Finished reading all BigQuery results") break } if err != nil { logger.Default().Error("Iterator Next failed", zap.Error(err)) errChan <- fmt.Errorf("it.Next: %w", err) return } outChan <- bqRanking } }() return outChan, errChan, nil } また、ハッシュタグ記事ランキングデータは直近5日間のデータのみを保管するという仕様があり、これまでは、DynamoDBのttlで自動削除していましたが、今回は、RDBに置き換えたため、バッチ処理のロジック側で対応しました。そして、一度に削除するデータ量が1.5GBほどで、一括削除するとCPU使用率が急上昇してしまうため、メトリクスを確認しながら、バッチサイズを調整し、逐次削除で対応しました。 ハッシュタグ記事ランキング取得APIの移行 CronJobのバッチ処理で書き込まれたハッシュタグ記事のランキングデータを、既存実装ではElasticBeanstalk上のAPIでクライアントに提供していましたが、ElasticBeanstalkを管理するawsアカウントは今後廃止する動きがあり、これらAPIも、Amebaブログのサブシステムが集まるEC2側に移行することが決まりました。 既存のDynamoDBに依存するハッシュタグ関連のAPIは全部で9個ほどありましたが、期間の都合上、今回は二つのAPIを移行しました。 実装に関しては、DynamoDBを利用したinfra層のロジックを理解した上で、Amebaブログで使われているbobというGoのORMを利用しましたが、従来のコードと比べると、シンプルかつ少ないコード量になることが実感できました。 旧実装(DynamoDBから取得) func (r *tagRankingRepositoryImplV2) listTagRankingEntriesByAmebaIDSimple( ctx context.Context, dt string, amebaID model.AmebaID, limit int, offsetRank *int, offsetEntryID *model.EntryID, offsetHashtag *model.HashtagID ) (es []blogtagRankingEntryV2, nextOffsetRank *int, nextOffsetEntryID *model.EntryID, nextOffsetHashtag *model.HashtagID, err error) { key := fmt.Sprintf("%s_%s", amebaID.String(), dt) var exclusiveStartKey map[string]types.AttributeValue if offsetRank != nil && offsetEntryID != nil && offsetHashtag != nil { exclusiveStartKey = map[string]types.AttributeValue{ "rank": &types.AttributeValueMemberN{ Value: fmt.Sprintf("%d", *offsetRank), }, "ameba_id_dt": &types.AttributeValueMemberS{ Value: key, }, "ameba_entry_id_tag_name": &types.AttributeValueMemberS{ Value: fmt.Sprintf("%s_%s_%s", amebaID.String(), offsetEntryID.String(), offsetHashtag.TagName()), }, "dt": &types.AttributeValueMemberS{ Value: dt, }, } } res, err := r.client.Query(ctx, &dynamodb.QueryInput{ TableName: &r.blogtagEntryRankingTable, IndexName: aws.String("ameba_id_dt-rank-index"), KeyConditionExpression: aws.String("ameba_id_dt = :ameba_id_dt"), FilterExpression: aws.String("official_tag_flag = :official_tag_flag"), ExpressionAttributeValues: map[string]types.AttributeValue{ ":ameba_id_dt": &types.AttributeValueMemberS{ Value: key, }, ":official_tag_flag": &types.AttributeValueMemberN{ Value: "1", }, }, Limit: aws.Int32(int32(limit)), ScanIndexForward: aws.Bool(true), ExclusiveStartKey: exclusiveStartKey, }) if err != nil { return nil, nil, nil, nil, fmt.Errorf("r.client.Query: %w", err) } if res.LastEvaluatedKey != nil { lastEvaluatedKey := res.LastEvaluatedKey var l blogtagRankingEntryV2 if err := attributevalue.UnmarshalMap(lastEvaluatedKey, &l); err != nil { return nil, nil, nil, nil, fmt.Errorf("attributevalue.UnmarshalMap: %w", err) } nextOffsetRank = &l.Rank eid := model.EntryID(l.EntryID()) nextOffsetEntryID = &eid hid := model.NewHashtagID(l.TagName()) nextOffsetHashtag = &hid } var blogtagRankingEntries []blogtagRankingEntryV2 if err := attributevalue.UnmarshalListOfMaps(res.Items, &blogtagRankingEntries); err != nil { return nil, nil, nil, nil, fmt.Errorf("attributevalue.UnmarshalListOfMaps: %w", err) } return blogtagRankingEntries, nextOffsetRank, nextOffsetEntryID, nextOffsetHashtag, nil } func (r *hashtagEntryRankingRepositoryImpl) listTagRankingEntriesByAmebaIDSimple( ctx context.Context, dt civil.Date, amebaID model.AmebaID, limit int, offsetRanking *int, offsetEntryID *model.HashtagEntryID, offsetHashtag *model.HashtagID, ) ([]hashtagdb.BlogHashtagEntryRanking, error) { qb := mysql.Select( sm.From(hashtagdb.BlogHashtagEntryRankings.Name()), sm.Where(hashtagdb.BlogHashtagEntryRankings.Columns.AmebaID.EQ(mysql.Arg(amebaID.String()))), sm.Where(hashtagdb.BlogHashtagEntryRankings.Columns.DT.EQ(mysql.Arg(dt.String()))), sm.Where(hashtagdb.BlogHashtagEntryRankings.Columns.OfficialTagFlag.EQ(mysql.Arg(1))), ) if offsetRanking != nil && offsetEntryID != nil && offsetHashtag != nil { qb.Apply(sm.Where(mysql.Raw( "(`rank`, entry_id, tag_name) > (?, ?, ?)", mysql.Arg(*offsetRanking), mysql.Arg(offsetEntryID.String()), mysql.Arg(offsetHashtag.TagName()), ))) } qb.Apply( sm.OrderBy(clause.OrderDef{ Expression: hashtagdb.BlogHashtagEntryRankings.Columns.Rank, Direction: " ASC", }), sm.OrderBy(clause.OrderDef{ Expression: hashtagdb.BlogHashtagEntryRankings.Columns.EntryID, Direction: " ASC", }), sm.OrderBy(clause.OrderDef{ Expression: hashtagdb.BlogHashtagEntryRankings.Columns.TagName, Direction: " ASC", }), sm.Limit(int64(limit)), ) rows, err := bob.All(ctx, r.replicaDBs.Next(), qb, scan.StructMapper[hashtagdb.BlogHashtagEntryRanking]()) if err != nil { if errors.Is(err, sql.ErrNoRows) { return []hashtagdb.BlogHashtagEntryRanking{}, nil } return nil, fmt.Errorf("bob.All: %w", err) } return rows, nil } 上記コードを比較すると、SQLでの実装が以下二つの観点で改善されている事がわかります。 SQLでの実装の方では、クエリの意図がコードから直接読み取れる一方で、DynamoDBはやや難解です。ExclusiveStartKey ameba_entry_id_tag_name amebaID_entryID_tagName ScanIndexForward: true DynamoDB版の実装を追うと、コスト効率の観点で問題があった事がわかります。FilterExpression official_tag_flag = 1 MySQLに置き換えたことにより、WHERE句とインデックスによって該当レコードのみを正確に読み取れるようになり、不要なレコードの読み取りが排除されました。これによりRead Capacityの無駄な消費がなくなり、コスト効率が改善されました。 この記事では、Amebaブログのハッシュタグ記事ランキングのストレージ移行とシステム刷新について紹介しました。今回の刷新により、インフラコストの削減と運用効率化が期待されます。

AmebaブログにおけるDynamoDBからMySQLへのストレージ移行の取り組み

背景や根拠まで確認しますか？

調べる

選ぶ

サイト