Apache DataFusionとは？Rust製SQLクエリエンジンの実行計画と導入手順【2026年9月版】

Apache DataFusionは、Rustで書かれた組み込み向けのSQLクエリエンジンです。「Apache Arrowプロジェクトの一部」という説明を見かけますが、その体制は2024年4月に終わっています。本記事は2026年9月時点の一次情報と、datafusion-python 54.0.0で実際に実行した出力をもとに、DataFusionの現在地と導入手順を整理します。

まとめ：Apache DataFusionの要点

DataFusionは「すぐ使える分析データベース」ではなく、自作のデータ基盤にSQL実行部分を移植するための部品です。単体で完結するツールとして評価すると、必ず機能不足に見えます。逆に、Parquetを読んでSQLを走らせる層を自前で書こうとしている開発者にとっては、クエリプランナと最適化器と並列実行器が丸ごと手に入る点が決定的な価値になります。

Apache Arrowの一部という説明が古い理由：2024年のトップレベル昇格

DataFusionは2019年にApache Software Foundationへ寄贈され、当時はコミュニティ規模が独立に足りなかったため、Apache Arrowプロジェクトのサブプロジェクトとして運営されていました。この体制は2024年に終わっています。公式ブログは次のように書いています。

as of April 16, 2024 the Apache Arrow DataFusion subproject is now a top level Apache Software Foundation project.（Announcing Apache Arrow DataFusion is now Apache DataFusion）

ASF理事会も2024年4月17日の会議でApache DataFusion Projectの設立決議を承認しています（ASF Board Minutes: DataFusion）。同時にプロジェクト名も Apache Arrow DataFusion から Apache DataFusion へ変わりました。ドキュメントのドメインも datafusion.apache.org に移っています。Arrowのメモリフォーマットを使う関係は続いていますが、ガバナンスもリリース番号もArrowとは別系統です。Arrowの版に合わせて選ぶ、という読み替えは成り立ちません。

公式の自己定義は次の一文です。

DataFusion is a very fast, extensible query engine for building high-quality data-centric systems in Rust, using the Apache Arrow in-memory format.（Introduction | Apache DataFusion documentation）

DataFusionは本体だけの単一プロジェクトではありません。公式サイトのトップページは、利用者向けに次の4つを関連プロジェクトとして紹介しています（ASFのリリースカタログにはこのほかsqlparser-rsも並びます）。名前が似ているうえリリース番号が揃っていないため、検索で拾った記事がどれを指しているのか取り違えやすい部分です。

なお、Google Cloudの「Cloud Data Fusion」は公式が powered by the open source project CDAP と説明する別サービスで、本記事の対象とは開発元も用途も重なりません。Rust製のほうを指すときは Apache を付けて表記するのが安全です。

実行計画に現れる3つの最適化：投影・フィルタのプッシュダウンと16並列

DataFusionが何をしているかは、EXPLAINを1回叩けば説明文より早くわかります。以下はdatafusion-python 54.0.0（Python 3.13）で、次のsales.csvに対して集約クエリの実行計画を出したものです。

id,region,product,amount,ts
1,east,alpha,1200,2026-09-01
2,west,beta,800,2026-09-01
3,east,beta,1500,2026-09-02
4,north,alpha,400,2026-09-02
5,west,alpha,2200,2026-09-03

from datafusion import SessionContext
ctx = SessionContext()
ctx.register_csv("sales", "sales.csv")
ctx.sql("EXPLAIN SELECT region, sum(amount) FROM sales WHERE amount > 1000 GROUP BY region").show()

+---------------+-------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                            |
+---------------+-------------------------------------------------------------------------------------------------+
| logical_plan  | Aggregate: groupBy=[[sales.region]], aggr=[[sum(sales.amount)]]                                 |
|               |   Filter: sales.amount > Int64(1000)                                                            |
|               |     TableScan: sales projection=[region, amount], partial_filters=[sales.amount > Int64(1000)]  |
| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[region@0 as region], aggr=[sum(sales.amount)]        |
|               |   RepartitionExec: partitioning=Hash([region@0], 16), input_partitions=16                       |
|               |     AggregateExec: mode=Partial, gby=[region@0 as region], aggr=[sum(sales.amount)]             |
|               |       FilterExec: amount@1 > 1000                                                               |
|               |         RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1                   |
|               |           DataSourceExec: file_groups={1 group: [[tmp/dftest/sales.csv]]},                      |
|               |             projection=[region, amount], file_type=csv, has_header=true                         |
+---------------+-------------------------------------------------------------------------------------------------+

この出力から3つのことが読み取れます。まず projection=[region, amount]。テーブルは5列ありますが、スキャンが上位へ渡す列は2列に絞られています。CSVのような行指向のテキストでは読み取り自体は行単位なので削減幅は限られますが、Parquetのように列単位で配置されたファイルでは、この投影がそのまま読み取りバイト数の削減になります。次に partial_filters。WHERE句が集約より下、スキャン直上まで降りています。最後に RoundRobinBatch(16) と Hash([region@0], 16)。入力1パーティションを16に割り直し、部分集約（mode=Partial）を並列に走らせてから最終集約（mode=FinalPartitioned）でまとめる、という2段構えの集約になっています。この16は実行環境の論理コア数です。

ではこの16並列が実際にどれだけ効いているのか。100万行のParquetを作り、target_partitions を変えて同じ集約を測りました。環境はMacBook Pro 16,1（Intel Core i9-9880H、論理16コア）、datafusion-python 54.0.0、Python 3.13、PyArrow 25.0.1です。

import time, pyarrow as pa, pyarrow.parquet as pq
from datafusion import SessionContext, SessionConfig

regions = ["east", "west", "north", "south"]
n = 1_000_000
pq.write_table(pa.table({
    "id":      pa.array(range(n), pa.int64()),
    "region":  pa.array([regions[i % 4] for i in range(n)]),
    "product": pa.array([f"p{i % 50}" for i in range(n)]),
    "amount":  pa.array([(i * 37) % 10000 for i in range(n)], pa.int64()),
}), "big.parquet", compression="snappy")        # 6,010,440バイト

q = "SELECT region, count(*) AS n, sum(amount) AS total FROM big GROUP BY region ORDER BY region"
for p in (1, 16):
    ctx = SessionContext(SessionConfig().with_target_partitions(p))
    ctx.register_parquet("big", "big.parquet")
    ctx.sql(q).collect()                        # ウォームアップ（計測外）
    ts = []
    for _ in range(5):
        t0 = time.perf_counter(); ctx.sql(q).collect(); ts.append(time.perf_counter() - t0)
    ts.sort()
    print(f"target_partitions={p}: 中央値 {ts[2]:.3f} s")

100万行程度では、並列度を16から1に落としても中央値はほとんど変わりません。むしろ16並列側は最小0.060秒から最大0.138秒までばらつき、1並列のほうが安定しています。実行計画に16並列と出ていても、この規模では並列化がボトルネックの解消に効いていないということです。実行計画の見た目を性能の根拠にしないでください。

上の表はウォームアップ後の値です。別途、既定設定のまま新しいプロセスで1回だけ実行したときは0.483秒かかりました。Parquetのメタデータ読み取りとスレッドプールの立ち上げが初回に乗るためで、1回計測のベンチマークはこの分を含んでしまいます。集計結果は次のとおりです。

ctx.sql(q).show()     # 表示用の再実行。上の計測には含めていません

DataFrame()
+--------+--------+------------+
| region | n      | total      |
+--------+--------+------------+
| east   | 250000 | 1249500000 |
| north  | 250000 | 1250000000 |
| south  | 250000 | 1250250000 |
| west   | 250000 | 1249750000 |
+--------+--------+------------+

実行時間はCPUとデータ分布に大きく依存します。採用可否を判断するときは、この手順を自分のデータで一度回して基準値を作ってください。

導入は3経路：Rustクレート・Pythonパッケージ・datafusion-cli

Rustプロジェクトへの組み込みとMSRVの引き上げ

Cargo.tomlに依存を足すだけで使えます。非同期APIなので、Tokioのマルチスレッドランタイムを併記します。

[dependencies]
datafusion = "55.1"
tokio = { version = "1", features = ["rt-multi-thread"] }

注意すべきはRustの最低要求バージョンです。crates.ioのメタデータ上、54系までは rust-version が 1.88.0 でしたが、55.0.0（2026-08-18）で1.94.0へ引き上げられています。古いツールチェーンを固定しているCIでは55系に上げた瞬間にビルドが落ちるので、アップグレード前に rustc --version を確認してください。バージョンを上げられない事情があるなら、54.1.0（2026-07-21）で止めるのが現実的な選択です。CargoやRustのツールチェーン管理そのものに不安がある場合は、Cargo（Rust）とは？インストールと使い方・主要コマンド一覧を先に押さえておくと詰まりにくくなります。

Pythonから使う場合に起きる版ズレ

PyPIのパッケージ名は datafusion です。ここで見落としやすいのが、PythonバインディングはRust本体と別リポジトリで独立してリリースされるため、版番号も公開時期も一致しない点です。2026年9月16日時点ではRust本体が55.1.0、PyPIのdatafusionが54.0.0で、1メジャー分の差がありました。Rust側のリリースノートに載っている新機能が、Pythonからはまだ使えない期間があります。

さらに、Pythonが古いとエラーを出さずに古い版へ落ちます。54.0.0の requires_python は >=3.10 なので、pipは条件を満たす最後の版まで遡ります。macOS同梱のPython 3.9.6と、Python 3.13の仮想環境で同じコマンドを実行した結果が次のとおりです。

$ /usr/bin/python3 --version
Python 3.9.6
$ /usr/bin/python3 -m pip install --user datafusion
$ /usr/bin/python3 -c "import datafusion; print(datafusion.__version__)"
50.1.0

$ python3.13 -m venv dfv && ./dfv/bin/pip install datafusion
$ ./dfv/bin/python -c "import datafusion; print(datafusion.__version__)"
54.0.0

エラーは出ません。4メジャー古い50.1.0が入ったまま動きます。この50.1.0自体の要件はPython 3.9以上なので、3.8以前ではさらに別の版が選ばれるかインストール自体が失敗します。ドキュメントどおりに書いたコードが通らないときは、まず datafusion.__version__ を出力して、Pythonのバージョンを疑ってください。

datafusion-cliの導入

SQLを対話的に試すだけならCLIが最短です。cargoでソースからビルドする方法と、macOSならHomebrewで入れる方法があります。

cargo install datafusion-cli    # crates.io の datafusion-cli 55.1.0 をビルド
brew install datafusion         # Homebrew formula 名は datafusion（コマンドは datafusion-cli）

cargo版はDataFusion本体をまるごとコンパイルするため初回は数十分かかります。Homebrew formulaの名前は datafusion ですが、入るコマンド名は datafusion-cli です。CLIは起動したディレクトリを既定の探索パスにするので、対象ファイルのある場所で起動すれば SELECT * FROM 'data.csv'; のようにパスを直接書いてクエリできます。

CSV・Parquet・JSONを読んでSQLを実行する最小手順

公式のFeatures一覧は、追加プラグインなしで扱えるデータソース形式をCSV・Parquet・JSON・Avro・Arrowと記載しています。実際のAPIは形式ごとに register_*（テーブルとして登録）と read_*（DataFrameとして取得）が対になっており、Python版のSessionContextには register_csv / register_parquet / register_json / register_avro / register_arrow が揃っています。読み込み先はローカルファイルに限りません。公式ドキュメントはAWS S3・Azure Blob Storage・Google Cloud Storageからの非同期ストリーミングI/Oに対応すると明記しており、それ以外のストレージは ObjectStore トレイトの実装で足せます。

from datafusion import SessionContext

ctx = SessionContext()
ctx.register_csv("sales", "sales.csv")
ctx.sql("SELECT region, sum(amount) AS total FROM sales GROUP BY region ORDER BY total DESC").show()

# クエリ結果をそのままParquetへ書き出し、書いたものを読み直す
ctx.sql("SELECT * FROM sales").write_parquet("out_parquet")
ctx.register_parquet("sales_pq", "out_parquet")
ctx.sql("SELECT count(*) AS n, sum(amount) AS total FROM sales_pq").show()

DataFrame()
+--------+-------+
| region | total |
+--------+-------+
| west   | 3000  |
| east   | 2700  |
| north  | 400   |
+--------+-------+
DataFrame()
+---+-------+
| n | total |
+---+-------+
| 5 | 6100  |
+---+-------+

JSONは1行1レコードのNDJSON（改行区切りJSON）が前提で、この形式であること自体は譲れません。一方、レコード内の入れ子は扱えます。54.0.0で構造体フィールドの参照 user['name'] と配列の行展開 unnest(tags) が動くことを確認しました。取り込み前に平坦化するかどうかは、入力スキーマと投げたいクエリを見て決めれば十分です。

ctx.register_json("events", "events.json")   # {"id":1,"lvl":"warn"} が1行1件で並んだファイル
ctx.sql("SELECT lvl, count(*) AS n FROM events GROUP BY lvl ORDER BY lvl").show()

+-------+---+
| lvl   | n |
+-------+---+
| error | 1 |
| warn  | 2 |
+-------+---+

Parquetそのものの仕組みや、CSVから移行したときに何が変わるのかはParquetとは？読み方・仕組み・CSVとの違いをわかりやすく解説で整理しています。

つまずきやすい仕様：識別子の扱いとストレージ非搭載

引用符付き識別子の大文字・小文字の区別

DataFusionは引用符なしの識別子を小文字へ正規化し、二重引用符で囲んだ識別子は書いたとおりに照合します。つまり引用符を付けた瞬間、実際の列名と大文字・小文字まで一致していなければ弾かれます。他のDBから移してきたSQLで最も踏みやすい違いです。列名が小文字のテーブルと大文字のテーブルで、同じSQLがどう分かれるかを実測しました。

# 列名が小文字 region のテーブル
>>> ctx.sql('SELECT "REGION" FROM sales LIMIT 1').collect()
ValueError: Schema error: No field named "REGION".
Valid fields are sales.id, sales.region, sales.product, sales.amount, sales.ts.

# 列名が大文字 REGION のテーブル
>>> ctx.sql('SELECT REGION FROM upper').collect()
Schema error: No field named region. Valid fields are upper."ID", upper."REGION".

エラーメッセージが有効な列名を引用符付きで並べてくれるので、どちらのケースかはその場で判別できます。引用符を機械的に一括除去するのは逆効果です。大文字を含む列名や予約語では引用符がないと通らなくなります。移行時は引用符付き識別子を洗い出し、実スキーマと突き合わせて個別に直してください。

永続ストレージと汎用インデックスの不在

CREATE TABLEもDROP TABLEもSHOW TABLESも通りますし、インメモリのカタログにテーブルを登録して管理できます。持っていないのは、それを次回の起動まで残す層のほうです。SessionContextはプロセス内の状態にすぎず、落とせば登録もCREATE TABLEで作った表も消えます。B-treeのような汎用インデックスを張って点参照を速くする機構も標準では持ちません。この性質は欠陥ではなく設計方針で、ストレージ層を利用側に委ねているからこそ、時系列DBにもオブジェクトストレージ上のレイクハウスにも同じエンジンを載せられます。永続化されたデータベースが欲しい要件にそのまま当てると外します。そこはDuckDBか、後述の分散SQLエンジンの領分です。

show()の出力形式とDataFrame()行

datafusion-python 54.0.0でスクリプトから df.show() を呼ぶと、表の前に DataFrame() という行が出ます。戻り値は None なので、変数へ代入しても消えません。本記事の実行例に紛れている DataFrame() はこれです。show() は人が読むための整形出力なので、結果をプログラムで受け取るなら標準出力を解析せず collect() でArrowのRecordBatchとして取り出してください。表示は行数が多いと切り詰められますし、整形も将来変わり得ます。

DuckDB・Polars・Trinoとの使い分け

選び方は用途で割り切れます。手元のParquetやCSVをSQLで分析したいだけならDuckDBです。DataFusionがとくに有力になるのは、自分の作るシステムの中にSQL実行層を埋め込み、その挙動をカスタムしたいときです。独自のデータソースをTableProviderとして足す、独自の集約関数を登録する、最適化パスを差し込む、といった要求がないなら、DuckDBのほうが確実に早く着地します。DuckDBの守備範囲はDuckDBとは？特徴・用途とSQLite・PostgreSQLとの違いを解説で解説しています。

Polarsとは住み分けです。PolarsはDataFrame操作のAPI設計が中心でSQLは後付けのインターフェース、DataFusionはSQLの文法カバレッジとクエリプランナが中心にあります。Pythonから表形式データを手早く加工したいならPolars、SQLを受け付ける層を製品に持たせたいならDataFusionで外しません。なお両者がArrowを共有していても、片方のエンジンをもう片方から呼び出せるわけではありません。共通なのはメモリ上のデータ表現までです。

Trinoは規模の軸が違います。複数ノードにまたがる大規模なデータレイクやデータウェアハウスへ、既存のカタログ経由でSQLを投げる基盤です（公式サイトは exabyte scale data lakes という表現を使っています）。単一プロセスで完結するDataFusionとは競合しません。分散側の選定はTrinoとは？分散SQLクエリエンジンの仕組み・Presto/Sparkとの違い・導入を解説を参照してください。

採用リストの現在地：更新されているものと止まったもの

DataFusionの公式ドキュメントには Known Users という採用プロジェクト一覧があります。ただしこのページは「知っているプロジェクトがあればPRを送ってほしい」という自己申告制で、掲載後の状況変化には追随しません。定番の採用事例として引かれる名前のうち、いくつかはすでに動きが止まっています。2026年9月16日時点で一次情報を当たり直した結果が次の表です。

InfluxDB 3については、InfluxDataが自社構成をFDAPスタック（Apache Flight・DataFusion・Arrow・Parquet）と呼んで公表し、InfluxDB 3 CoreのリファレンスもSQL実装にDataFusionを使うと記載しています。時系列DBのSQL実行層をDataFusionに置き換えた、実運用の採用例です。

HoraeDBは公式のKnown Usersに今も名前が残っていますが、Apache Incubatorのステータスページは退役を明記しています。ASFの退役はコミュニティ活動の終了であってコードが消えることではありませんが、現在も開発が続く採用事例として紹介すれば誤解を招きます。引くなら過去の実績として、退役済みである旨を添えてください。dask-sqlも同様で、日本語の解説記事では定番の事例ですが、パッケージもリポジトリも2024年で更新が止まっており、現在のKnown Usersからは外れています。

一方で、動いているものは大きく動いています。Sparkのネイティブ実行プラグインであるComet は2026年8月7日に1.0.0へ到達しました。導入時に効く制約が2つあります。ひとつは配布形態で、Maven Centralに公開されているjarがネイティブライブラリを同梱しているのはLinux（amd64とarm64）だけです。Apple Siliconのローカル環境で試すにはソースからのビルドが要ります。もうひとつは対応Sparkで、CIでテストされているのは3.4.3・3.5.9・4.0.4・4.1.3の4系統です。1.0.0のドキュメントには次の警告が入っています。

JDK 11 and Spark 3.4 support are deprecated as of the 1.0.0 release and will be removed in the 1.1.0 release.（Installing DataFusion Comet。リリース告知はApache DataFusion Comet 1.0.0 Release）

分散実行では名前の近い2つを区別してください。Ballista はASFのDataFusionサブプロジェクト（apache/datafusion-ballista、crates.ioの ballista は54.1.0）で、DataFusion本体のバージョンに追随しています。2024年ごろの英語記事にある「現在は活発にメンテナンスされていない」という記述は現状を表しません。crates.ioの公開履歴では0.12.0（2024-02-07）のあと43.0.0（2025-01-20）でDataFusion本体と同じ番号体系に切り替わり、以降は54.1.0まで継続的にリリースされています。対して datafusion-distributed はASF外のコミュニティプロジェクト（datafusion-contrib/datafusion-distributed、crates.ioで4.0.0）で、DataFusionに分散実行の機能を後付けするライブラリとして別系統で開発されています。ガバナンスもリリース番号も別物なので、社内で採用可否を審査する際は取り違えないようにしてください。

よくある質問

Apache DataFusionはApache Arrowの一部ですか？

いいえ。2024年4月16日にApache Arrowのサブプロジェクトからトップレベルプロジェクトへ独立しました。Arrowのインメモリフォーマットを使う関係は続いていますが、リリース番号もドキュメントサイトも別系統です。

Google CloudのData Fusionと同じものですか？

別物です。Cloud Data FusionはGoogle CloudのGUIベースETLサービスで、Apache DataFusionはRust製の組み込み向けSQLクエリエンジンです。開発元も用途も重なりません。

datafusion-cliはどうやってインストールしますか？

Rust環境があれば cargo install datafusion-cli でcrates.ioの55.1.0をビルドできます。macOSなら brew install datafusion でも導入でき、どちらの場合もコマンド名は datafusion-cli です。cargo経由は本体をフルビルドするため初回に時間がかかります。

Python版とRust版でバージョン番号が違うのはなぜですか？

Pythonバインディングは別リポジトリで独立してリリースされるため、本体より遅れます。2026年9月16日時点でRust本体は55.1.0、PyPIのdatafusionは54.0.0でした。Python 3.9以下ではさらに古い50.1.0がエラーなしで入るので、動作がドキュメントと合わないときは datafusion.__version__ を確認してください。

DataFusionとDuckDBはどちらを選ぶべきですか？

手元のParquetやCSVをSQLで分析する目的ならDuckDBです。DataFusionがとくに有力になるのは、自作システムにSQL実行層を組み込み、データソースや関数や最適化パスを自分で拡張したいときです。DataFusionは独自の永続ストレージを持たないため、データを保存しておくデータベース製品の代替にはなりません。

関連記事

• DuckDBとは？特徴・用途とSQLite・PostgreSQLとの違いを解説
• Trinoとは？分散SQLクエリエンジンの仕組み・Presto/Sparkとの違い・導入を解説
• Parquetとは？読み方・仕組み・CSVとの違いをわかりやすく解説
• Cargo（Rust）とは？インストールと使い方・主要コマンド一覧