Cloudflare R2にAuto Loaderを向けてみたら、こうなった！？

お疲れ様です、桑野です！

R2でAutoloader結論から言うと動きました！ただ、制約はありますｗ
加えてこれはDatabricksの公式見解ではなく、自分が試した結果となります！

Cloudflare R2にログを置いている人、けっこういると思います！
ただR2からDatabricksに持ってこれるかについて調べてみたかったので実際にAuto Loaderが使えるか試してみました。

とりあえずexternal locationを作ろ

まずR2のバケットをUnity Catalogから見えるようにします。用意するものは2つで、R2のAPIトークンを入れたstorage credentialと、それを使うexternal locationです。

パスの形がちょっと独特で、r2://<バケット名>@<アカウントID>.r2.cloudflarestorage.com/<パス> になります。S3の s3:// に慣れているとアカウントIDのところがちょいめんどいっすねｗ

ここは普通に作れると思います！ここが作れなかったら大変ですｗ

DBR 14.3以降か、SQL warehouseなら2024.15以降が必要です！（もうそんな古いDBRは中々使ってないと思いますが）

こうやってテストしたんよ

テストデータの置き方は、R2のパスに外部ボリュームを作れば、databricks fs cp でそのまま書き込めます。CLIだけで完結するので、ファイルを置くのがだいぶ楽です。

置くのは1行1レコードのJSONです。増分が効いているかは、各行に取り込み時刻を持たせて、バッチごとに時刻のdistinctを数えれば分かります。段取りは最初に3ファイル、あとから2ファイル追加してみました！

Directory Listingはあっさり動いた

拍子抜けするくらい、あっさり動いてくれました！

書いたのはこれだけ！

			
spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaHints", SCHEMA_HINTS)
    .option("cloudFiles.includeExistingFiles", "true")
    .load(EVENTS_PATH)

		

1回目で15行・3ファイルが入りました。同じパスを read_files() でバッチ読みしても15行なので、数は合っています！

そこに2ファイル足して回し直すと +8行で合計23行。count(*) と count(DISTINCT event_id) がどちらも23なので重複はゼロ、取り込み時刻のdistinctも各バッチ1回だけ。1回目のファイルを読み直してはいないということ！
実行はLakeflow Spark Declarative Pipeline、channelは CURRENT です。

ここまでは何も困りませんでした。なんでかというと、Directory ListingモードってS3のコマンドしか使ってないんすよ。R2はS3のコマンド完全互換を謳っているので問題なく実行できるはず、と思ってました。

計算通り（AA略

notificationモードで壁に当たった

本番はここからだ！と cloudFiles.useNotifications をtrueにしたら、起動した瞬間に落ちましたｗ

			
[CF_EVENT_NOTIFICATION_NOT_SUPPORTED] Auto Loader event notification mode is not
supported for custom FileSystem. SQLSTATE: 0A000

custom FileSystem、つまりDatabricksから見て「独自のファイルシステム」の扱いだと言われています。S3ならSNSとSQSでファイルの到着を通知してもらえますが、R2にはそれに相当するものが無いわけです。

じゃあexternal location側のmanaged file eventsならどうか。こちらも弾かれました。

			
$ databricks external-locations update r2-test --json '{"enable_file_events": true}'
Error: UpdateExternalLocation must provide a file event queue

キューを出せと言われても、出せるキューがございませんｗ　effective_enable_file_events も false のままです。設定を直せば済む話ではなく、構造としてそうなっています。

ということで、File Notificationモードについては動かない、と言う結果になります、これはクラウド環境に依存するモードなので仕方ないと思います。

結論です

つまりR2に対しては、Directory Listingしか選べません。

これは「ファイルが増えたときのリスト処理がそのまま重くなる」ということでもあって、基本的にFile Notificationモードをオススメすることが多いのですが、使えないのでどうするか、、、。

じゃあどうするか。現時点では設計で逃げるしか無く、日付でディレクトリを掘って起点のパスを絞る、pathGlobFilter で対象を狭める、取り込み済みは cloudFiles.cleanSource で別のプレフィックスへ逃がす。フラットな1ディレクトリに大量のオブジェクトを置かない、などです。
最初から前提に入れておけば、そんなに困らないはずです。

R2に置いたデータをDatabricksから見たい方は、動くこと自体は確認できたのでぜひ試してみてください！

ではでは、今日は小ネタでした！

コメントを残すコメントをキャンセル