導入

以下のDatabricks公式Blogにて、バリアントデータ型の導入が紹介されていました。

Delta 4.0/Spark 4.0からバリアントサポートは含まれるようであり、DatabricksではDBR 15.3より利用可能になるとのこと。

既にDatabricks社員の方が試された記事も掲載されていました。

DBRは15.3がベータとして公開されていることもあり、私も興味津々でしたので、少し試してみました。

検証環境はDatabricks on AWS、DBRは15.3ML(Beta)です。

上記の通りDBR15.3はベータリリースのため、将来的に下記コードは動かなくなる可能性があります。

とりあえず動かす

Pythonのノートブックを作って、JSON形式の文字列を含んだデータフレームを作成します。

意図的に異なる構造のデータにしたかったので、以下のように2種のJSON形式文字列を定義。

json_str1 = '{"name":"Bob", "age":30, "cars": ["Ford", "BMW"], "pets": null}'
json_str2 = '{"id":"1234567890", "name":"Jhon Doe", "first_name":"Jhon", "last_name":"Doe", "address": {"street":"Main St", "city":"New York", "state":"NY"}}'

作成したJSON型文字列を使ってデータフレームを作成。

df = spark.createDataFrame(
    [["001", json_str1], ["002", json_str2]],
    ["id", "json_str"],
)

# SQLで利用するため、Temporary Viewとしても定義
df.createOrReplaceTempView("json_table")
display(df)

以下のようなデータフレームができました。

ちなみに、アイコンを見て分かるように、全ての列は文字列型です。

念のため、スキーマ情報を出力。

df.printSchema()

出力

root
 |-- id: string (nullable = true)
 |-- json_str: string (nullable = true)

文字列ですね。

では、上記のブログにあるように、SQLでparse_jsonを使ってバリアント型へ変換してみます。

%sql
SELECT
  id,
  parse_json(json_str)
FROM
  json_table;

JSON文字列がバリアント型へ変換されました。
また、異なる構造のJSONを同一列に含めましたが、問題なく変換されました。

従来は、構造の異なるJSON文字列を非文字列型データとして保持する際は、スキーマを合わせてStruct型で保管するか、フラットな構造の場合はDict型で保管するようなやり方だったかなと思います。
データによってはこういった変換や管理は手間でしたが、バリアント型を使えるようになることでより柔軟に半構造化データ列を管理できるようになりますね。

Pysparkで使う

SQLではなくPython(Pyspark)でもやってみましょう。

Pysparkでもparse_json関数が提供されていますので、それを使って変換します。

import pyspark.sql.functions as F

variant_df = df.select("id", F.parse_json("json_str").alias("json_obj"))

display(variant_df)

SQLと同等の結果が得られました。

では、Variant列から要素を取り出すクエリを実行してみます。
SQLライクにパスナビゲーションを書く場合は以下のようになります。

variant_df.selectExpr("id", "json_obj:name::string").display()

パスナビゲーションは列名:ドット区切りの要素名::変換後データタイプという書式の模様。

したがって、ネストした要素へのアクセスは以下のようにドット区切りで指定します。

variant_df.selectExpr("id", "json_obj:address.city::string").display()

なお、変換後データタイプ指定を省略すると、バリアント型のデータとして結果が得られます。

また、変換不能なデータタイプを指定すると、エラーになります。

# INT型への変換はできない
variant_df.selectExpr("id", "json_obj:address.city::int").display()

よりPythonライクな記述をしたい場合は、variant_get関数を使うことが出来ます。

variant_df.select("id", F.variant_get("json_obj", "$.name", "string")).display()

variant_getは第一引数にバリアント型の列名、第2引数に要素のパスナビゲーション、第3引数に変換後データタイプを指定する関数です。詳細は以下を確認ください。

variant_getもデータタイプに変換不可能なものを指定すると、エラーになります。

このようなエラーを避けたい場合、variant_getの代わりにtry_variant_getを利用することができます。

variant_df.select("id", F.try_variant_get("json_obj", "$.name", "int")).display()

try_variant_getはエラーが出ない代わりに、変換不能の場合は結果がnullとなります。

最後に、バリアント型列を含むデータフレームをDelta Tableとして保管します。

# Delta Tableへ保管
variant_df.write.mode("overwrite").saveAsTable("training.default.variant_sample")

無事にできました。

カタログエクスプローラ上でのプレビューも問題なく出来ます。

詳細を見ると、Propertiesとしてdelta.feature.variantType-preview=supportedが設定されていました。(今だけかな)

まとめ

JSON形式で半構造データを取り扱う場合、かなりやり易くなったという印象です。
(スキーマを厳密に管理すべきユースケースにおいては注意が必要ですが)

最近だとLLMでJSON形式の構造化出力を行うことが増えてきそうだなと感じているのですが、そいういった出力結果の管理や利用の利便性向上につながるのではないかと期待しています。

かなり汎用的に使えそうな機能ですので、広範なユースケースでの適用が今後も増えて行きそうですね。

Delta LakeとSparkのバリアントデータ型をDatabricksで試す

導入

とりあえず動かす

Pysparkで使う

まとめ