MaxCompute での小さいファイルのマージ - MaxCompute - Alibaba Cloud ドキュメントセンター

分散ファイルシステムは、データをブロック単位で保存します。64 MB のブロックサイズより小さいファイルは、小さいファイルと見なされます。分散システムでは、SQL タスクの出力、他の分散エンジン、または Tunnel を介して収集されたデータなどのソースから、必然的に小さいファイルが生成されます。これらの小さいファイルをマージすることで、システムのパフォーマンスが最適化されます。このトピックでは、MaxCompute で小さいファイルをマージする方法について説明します。

背景情報

小さいファイルが多すぎると、以下の問題が発生する可能性があります。

MaxCompute は、複数の小さいファイルを処理するよりも、単一の大きいファイルを処理する方が効率的です。小さいファイルが多すぎると、全体的な実行パフォーマンスが低下する可能性があります。
小さいファイルが多すぎると、Pangu ファイルシステムに負荷がかかり、ストレージ効率が低下する可能性があります。

そのため、ストレージとパフォーマンスの両方を最適化するには、計算中に生成された小さいファイルをマージすることが重要です。MaxCompute には、小さいファイルを処理するための強力な機能がいくつかあります。

デフォルトでは、タスク完了後、特定の条件を満たすと、システムが自動的に Fuxi タスクを割り当てて小さいファイルをマージします。このプロセスは MergeTask と呼ばれます。
デフォルトでは、Fuxi インスタンスは最大 100 個の小さいファイルを処理できます。odps.sql.mapper.merge.limit.size パラメーターは、読み取るファイルの合計サイズを制御します。
MaxCompute は、バックグラウンドで定期的にメタデータベースをスキャンし、小さいファイルを多く含むテーブルまたはパーティションの小さいファイルをマージします。このプロセスはユーザーに対して透過的です。

しかし、メタデータの検出により、マージされていない小さいファイルが多数見つかることがあります。たとえば、テーブルに継続的に書き込みが行われている場合、自動マージ操作は実行できません。この場合、まず書き込みタスクを停止してから、手動で小さいファイルをマージする必要があります。

注意事項

小さいファイルのマージは、コンピューティングリソースを消費します。お使いのインスタンスが従量課金制の場合、この操作は課金対象となります。課金ルールは、従量課金の SQL タスクと同じです。詳細については、「コンピューティングの料金 (従量課金)」をご参照ください。
小さいファイルをマージするコマンドは、トランザクションテーブルをサポートしていません。トランザクションテーブルの小さいファイルをマージするには、コンパクションをご参照ください。

テーブル内のファイル数の確認

構文

DESC EXTENDED <table_name> [PARTITION (<pt_spec>)];

パラメーター
- table_name
  必須。確認するテーブルの名前。
- pt_spec
  任意。チェックするパーティションで、形式は (partition_col1 = partition_col_value1, partition_col2 = partition_col_value2, ...) です。

例

odl_bpm_wfc_task_log テーブルのサンプルパーティションには 3,607 個のファイルがあります。パーティションサイズは 274 MB (287,869,658 バイト) で、平均ファイルサイズは約 0.07 MB です。このパーティション内の小さいファイルをマージする必要があります。

odps@ admin_task_project>desc extended jxbwarehouse.odl_bpm_wfc_task_log partition(new_dts_sync_modifytime_year='2022',new_dts_sync_modifytime_month='11',new_dts_sync_modifytime_day='27');
+--------------------------------------------------------------------------------------------+
| PartitionSize: 287869658                                                                   |
+--------------------------------------------------------------------------------------------+
| CreateTime:               2022-11-27 00:00:46                                              |
| LastDDLTime:              2022-11-27 00:00:46                                              |
| LastModifiedTime:         2022-11-27 23:59:47                                              |
+--------------------------------------------------------------------------------------------+
| IsExstore:                false                                                            |
| IsArchived:               false                                                            |
| PhysicalSize:             863608974                                                        |
| FileNum:                  3607                                                             |
+--------------------------------------------------------------------------------------------+

ソリューション

メタデータ検出により、ファイルのマージに適したパーティションを特定できます。パーティションに 100 を超えるファイルが含まれ、平均ファイルサイズが 64 MB 未満の場合、マージを推奨します。解決策は 2 つあります。

即時マージ
次のコマンドを使用して、即時マージを実行できます。
```
ALTER TABLE <table_name> [partition (<pt_spec>)] MERGE SMALLFILES;
```
odl_bpm_wfc_task_log テーブルで即時マージ操作を実行すると、ファイル数は 19 に、ストレージサイズは 37 MB に削減されます。このことから、小さいファイルをマージするとストレージ効率も大幅に向上することがわかります。
ほとんどの場合、デフォルトのパラメーターで小さいファイルをマージするには十分です。ただし、MaxCompute には、カスタム要件を満たすためのパラメーターがいくつか用意されています。主なパラメーターは次のとおりです。
- SET odps.merge.cross.paths=true|false
  パスをまたいでファイルをマージするかどうかを指定します。テーブルに複数のパーティションがある場合、マージプロセスはパーティションごとに個別の MergeAction を生成します。したがって、odps.merge.cross.paths を true に設定しても、パスの数は変わりません。各パス内の小さいファイルが個別にマージされるだけです。
- SET odps.merge.smallfile.filesize.threshold = 64
  マージされる小さなファイルのサイズのしきい値を MB 単位で指定します。このしきい値より大きいファイルはマージされません。このパラメーターを設定する必要はありません。このパラメーターが設定されていない場合、デフォルト値が 32 MB のグローバル変数 odps_g_merge_filesize_threshold が使用されます。このパラメーターを設定する場合、その値は 32 MB より大きい必要があります。
- SET odps.merge.maxmerged.filesize.threshold = 500
  マージ後の出力ファイルのサイズを MB 単位で指定します。出力ファイルがこのしきい値を超えた場合、新しい出力ファイルが作成されます。このパラメーターを設定する必要はありません。このパラメーターが設定されていない場合、デフォルト値が 500 MB のグローバル変数 odps_g_max_merged_filesize_threshold が使用されます。このパラメーターを設定する場合、その値は 500 MB より大きい必要があります。

PyODPS スクリプトを使用したマージ

PyODPS を使用して、前日のタスクで生成された小さいファイルをマージするタスクを非同期に送信できます。次のコードは、スクリプトの例です。

import os
from odps import ODPS

# ALIBABA_CLOUD_ACCESS_KEY_ID 環境変数に、ご自身のアクセスキー ID が設定されていることを確認します。
# ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数に、ご自身のアクセスキーシークレットが設定されていることを確認します。
# アクセスキー ID とアクセスキーシークレットの文字列を直接使用しないでください。
o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-endpoint',
)

# 注: $table_name をご自身のテーブル名に置き換えてください。
table_name = $table_name
t = o.get_table(table_name)

# マージオプションを設定します。
hints = {'odps.merge.maxmerged.filesize.threshold': 256}

# 複数パーティションの例
insts = []
# iterate_partitions は、ds=$datetime 配下のすべてのパーティションをリストアップします。 パーティションの形式は異なる場合があります。
for partition in t.iterate_partitions(spec="ds=%s" % $datetime):
    instance=o.run_merge_files(table_name, str(partition), hints=hints)
		# この Logview の [Waiting Queue] をクリックして、実際の実行の Logview を見つけます。
    print(instance.get_logview_address())
    insts.append(instance)

# パーティションの結果を待ちます。
for inst in insts:
    inst.wait_for_completion()

このスクリプトを実行するには、まず PyODPS をインストールする必要があります。インストール手順については、「PyODPS」をご参照ください。

ユースケース

tbcdm.dwd_tb_log_pv_diは、データ安定性システムが小さいファイルのマージが必要だと判断した物理テーブルです。メタデータテーブル tbcdm.dws_rmd_merge_task_1d によると、このテーブルのほとんどのパーティションには 1,000 個を超えるファイルが含まれており (一部は 7,000 個を超えるものもあります)、平均ファイルサイズは 1 MB 未満です。

次のコマンドを使用して、即時マージによって小さいファイルをマージできます。

SET odps.merge.cross.paths=true;
SET odps.merge.smallfile.filesize.threshold=128;
SET odps.merge.max.filenumber.per.instance = 2000;
ALTER TABLE tbcdm.dwd_tb_log_pv_di PARTITION (ds='20151116') MERGE smallfiles;