formatと命名する技術に対する関心が高まっています。Formatが管理が難しかったり、技術的に要件が満たされない(Update、Delete、ACID)など様々な理由から脚光を浴びています。Lake)がありますが、その中でHudiの性能最適化に関連したヒントを聞きたいと思い、このセッションを申し込みました。セッションの概要紹介このセッションはHands-On方式で行われましたが、時間的な制約のため、最適化されていないテーブルのテストは正確に行われませんでした。先行してテストしていただいた結果を共有してくださったので、その指標を一緒に確認しながら進行しました。サンプルデータはドローンデータを想定して作業を進めました。ドローンデータのTimestamp形式はyyyyMMddHHmmとなっており、当該ノートブックでは最適化されていない場合に下記のように表示されます。当該テストを進行するのに使われたEMRは6.12バージョンであり、Hudiはv0.13.1で進行しました。node)は160vCPUが満たされる基準で作業しました。sessionの設定は次のように進めたし、’spark.sql.shuffle.partitions’を使ってファイルサイズを調整する方式です(もちろん、他の設定値もありますが、一般的におすすめする一番簡単な方式だそうです)。ソースデータはCSV形式のデータを読み込んで、スキーマの構造は次のような感じです(この記事ではノートブックソースの詳細な部分は載せず、結果中心で共有したいと思います)。Icebergと違ってHudiには数多くの設定値が存在します。このような設定値をうまく扱うことができれば、状況によっては非常に良い性能を発揮します。パーカーの基本的なパラメータ値を確認すると、以下のような様々なパラメータ値が存在することが分かります(あくまで基本です!)。ひとつひとつ簡単に説明すると、次のように説明することができます。ディーはCOW(COPY_ON_WRITE)、MOR(MERGE_ON_READ)二つの方法で保存することができます。ここで設定されたCOWを説明すると、例えば、Aバージョンのデータファイルに(a,b,c)カラムが含まれていて、updateやdeleteの過程でそのデータが変わる場合、Aバージョンのファイルをコピーして変更された内容をBバージョンのファイルに反映して管理する方法を意味します。formatを使ってる方は知ってるかもしれません。物理的にパーティショニングを分けて管理することを検索やデータ処理にスピードを上げることができる最も簡単な方法です(例えば、2023年1月のデータが2023年フォルダに入ってるより2023年1月フォルダに管理される方が検索しやすいのと同じです)。このフィールドはcombineからヒントを得てください。keyがまるでRDBMSのPKと同じように動作します。fieldを基準に最新の値を判断して保存するため、Timestampあるいは時間を基準に指定することになります。partitioningはパーティショニングをするフォルダ名の構造をどのように取るかについての設定だと思えばいいと思います。styleと定義します。metastoreとsyncをするかどうかの設定です。metastoreとして使うので、イメージと同じように設定すると思います。Keyはイメージにも明示されていますが、uniqueである必要があります。これはRDBMSでよく使われる概念であるPKをS3のようなファイルシステムにも同じように適用できることを意味します(ただし、Indexポリシーをこのような方法であまりにもタイトにすると、当然、使用する時間において時間の複雑度が非常に高くなる可能性が高くなるので、適用には注意が必要です)。次はsortingの設定値です。Hudiではファイルを保存する時sortingを通してファイルサイズを冒頭で言及したshuffle.parallelismに基づいて調整することになり、これはファイルのサイズが一定に変更され、そのファイルのsortモードがGLOBAL(グローバル)的に発生することを意味します。次はindexポリシーです。keyをどのような方法で管理するかについての設定です。Indexingを確認してみてください。keyが存在しますが、ファイル全体を基準にはならないこともあります)。基本的に現在のオープンソースの3つのテーブル(?)の構造はmetadataがないと使えない構造になっています。これはファイルを実際変更するのではなく、新しいファイルを作ってバージョンを管理する方法で使ってるからだと理解してください。filterの値を追加してIndex基準で検索速度を上げます。これはオープンソースファイルフォーマットが実際検索をする時、全体のデータを検索するのではなく、メタデータベースで検索をするため、検索速度が速くなることができます。bloomを検索する時、そのメタデータを使って検索するかどうかの設定値です(上の設定値で使用することができます)。tableに追加的な情報を入力することです。このような色んな設定値を適用したS3ファイルの結果を見てみましょう。ご覧の通り、ファイルが約~40MBの大きさにチューニングされたことが確認できます。すると、155個のファイルが~20MB程度に小さく分割される結果が出ると思います。record_key基準で5つのファイルを選んで確認してみました。画像にも見えますが、record_keyを基準にファイルごとに重複したデータがなく、きれいにソートされていることを確認することができます。検索性能がどう変わるか確認してみましょう。検索はrecode_keyを基準にwhere条件節を与えた時、検索される結果が表示されます。テスト結果7.32sくらいかかりましたし、チューニングされてないテーブルを基準には40秒かかりました。それでは、このような結果がどのように出たのか詳しく確認してみましょう。parquet段階で読むファイルの数が155個vs1個で本当に圧倒的に減ったことが確認できます。追加的にColumnarToRow段階も一緒に減ったことが確認できます。コネクションを使う方法は下記の通りです(例示の方法はsnowflakeでBigquery,DB2などソースによって使い方が変わることがあります。)次のコードを通してGlue環境で私たちは簡単にsnowflakeのデータソースを取得してglueで分析することができます。query基準にするとどんな結果が出るでしょうか?上の基準で検索をした時、最適化されたテーブルは3~4sの時間で結果が出ましたが、最適化されてないテーブルは7sの時間がかかりました。これはデータをスキップする過程で発生する最適化で、当然、ファイルが多くてデータのサイズが大きいほど差が大きくなります。indexを適用したupsertの最適化結果を確認してみましょう。uiを確認してみると、様々な部分で一度に使えるTaskを同時に活用することで、時間が短くなることが確認できます。実際、このような作業の場合、最適化されたテーブルは6分30秒かかった一方、最適化されてないテーブルは8分36秒以上かかることが確認できます。skippingに対する技術を適用して、これによる効果がどの程度発生するかを確認することができました。これは大量のデータをS3及びデータレイクを管理する上で多くの時間的工数を減らしてくれて、また、検索性能及び前処理性能の向上で成り立って非常にポジティブな効果を与えると期待しています。このセッションで行ったワークショップは後日公開的にオープンするそうです。追加情報(もし、オープンテーブルフォーマットについてもっと知りたい方は、下記のリンクを参考してください。