DXC Data&AIPublicationへの投稿
🔎

DatabricksでJSONデータの読込、加工を行ってみた

tachi
に公開
初心者
JSON
Databricks
tech

初めに

データ処理では、構造化データと半構造化データを共に扱う場面もあります。DatabricksやSnowflakeなどのデータプラットフォームでは、半構造化データを扱うための機能が提供されています。
本記事では、Databricksを使って半構造化データの処理を実際に行ってみた際のコードや結果を紹介していきます。

Databricksでは複数のタイプのJSONデータを読み込んで処理することができます。
今回は、

  • 通常のJSONファイル（1つのJSONオブジェクトが複数行で記載されているもの）
  • JSON Lines形式のデータ（1行に1オブジェクトを書きこんだ形式）
  • JSON文字列（string型として保存されたJSONデータ）

の3種類のそれぞれを取り込み、Databricksで操作する実践の様子をご紹介します。
また、DatabricksではパブリックプレビューとしてVARIANT型として半構造化データを格納する機能を提供しています（2025年6月26日現在）。GA前の機能ですが、実践の最後にはこちらを使用したデータの取り込みの様子もご紹介します。

ファイルからのデータの取り込み

複数行モードでの読み込み

こちらで読み込むのは、以下のように、複数行で一つのオブジェクトを記述したファイルです。読み込むファイルは、DBFSにアップロードしました。

Python
df_1 = spark.read.option("multiline", "true").format("json").load("dbfs:/FileStore/tables/JSON.json")



キーと値がペアとなった形でJSONファイルからデータを取り込むことが出来ました。

一行一オブジェクトの場合の読み込み

ここで使用するのは、JSON Lines形式で記載したファイルです。（今回は拡張子は.jsonでファイルを作成しています。）

Python
df_1_5 = spark.read.format("json").load("dbfs:/FileStore/tables/JSON1_5.json")


一行一オブジェクトになっている場合、option("multiline", "true")が必要ありません。

CSV内のJSON文字列

データのロード

CSVファイル内にJSON文字列が格納されている場合のデータの読み込みを行います。
ここで使用するのは、以下のように、memberというカラムに文字列としてJSONオブジェクトが格納されたCSVファイルです。

Python
df_4 = spark.read.format("csv").option("header", "true").option("inferSchema", "true").option("multiLine", "true").option("escape", "\"").option("quote", "\"").load("dbfs:/FileStore/tables/JSON4.csv")


CSVファイルを読み込みました。
この段階では、まだmemberカラムはstring型です。

JSON文字列をスキーマで定義した形の構造体に変換する

読み込んだ段階では、string型として保存されるため、スキーマを定義してJSON文字列部分をstruct型に変換します。

Python
from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType,StringType, ArrayType, IntegerType
member_schema = StructType().add("name", StringType()).add("age", IntegerType()).add("city", StringType()).add("email", ArrayType(StringType())).add("phone", StringType())
df_4_parsed = df_4.withColumn("member_struct", from_json(df_4.member, member_schema))

struct型に変換したことで、上記の「複数行モードでの読み込み」や「一行一オブジェクトの場合の読み込み」で読み込んだデータと同様に「struct型のカラム名.キー名」を用いて値を取り出すことが出来ます。

データの操作

続いて、データの操作の例として、JSON内のキーの名前やデータ型の変更を行ってみます。
使用するのは、配列やネストされたオブジェクトを含む複雑な構造を持つデータです。
1つのオブジェクトの中に、memberというオブジェクトの配列が格納される構造となっています。

まず、ファイルからデータを読み込みます。

Python
df_3 = spark.read.option("multiline", "true").format("json").load("dbfs:/FileStore/tables/JSON3.json")

string型のfamily nameと、struct型の要素の配列であるmemberとして読み込むことが出来ました。

データの加工

カラム名の変更と、ネストされたオブジェクト内の値の形式の変更を行います。

まず、struct型内部の項目の操作を行うために、データの構造を変形していきます。
string型のカラム名を変更（family name→family_name）し、配列に格納されているオブジェクトを1レコード1オブジェクトに展開します。

Python
df_3_re_exp = df_3.select(col("family name").alias("family_name"), explode(col("member")).alias("member"))


次に、struct型内のカラムのリネーム（name→full_name、city→location）と、array型であるemailカラムの要素の結合、文字列化を行います。

Python
from pyspark.sql.functions import concat_ws
renamed_df = df_3_re_exp.select(col("family_name"), col("member.name").alias("full_name"), col("member.age"), col("member.city").alias("location"), col("member.email"), col("member.phone")).withColumn("emails", concat_ws(" ", "email")).drop("email")


カラムを一つずつセレクトしているため、struct型ではなくなっています。

データの構造を操作前の状態に戻す

フラット化したカラムをstruct型に戻します

Python
resturuct_df = renamed_df.select(col("family_name"),struct(("full_name"), col("age"), col("location"), col("phone"), col("emails")).alias("member"))

展開したmemberを、配列に戻します。

Python
grouped_df = resturuct_df.groupBy("family_name").agg(collect_list("member").alias("member"))

ネストされたJSON内のキー名とデータの操作・データ型の変更を行い、カラム名
・値の操作前の構造に戻すことが出来ました。

VARIANT型

VARIANT型でもJSONデータの操作を行ってみます。ここでは、「CSV内のJSON文字列」で使用したファイルを読み込み、カラム名を一部変更（family name→family_name）して保存したDeltaテーブルを使用しています。

Python
df_4_rename= df_4.withColumnRenamed("family name", "family_name")
df_4_rename.write.format("delta").mode("overwrite").saveAsTable("family_json")

VARIANT型として読み込み、テーブルを作成

Deltaテーブルを読み込みます。memberカラムのJSON文字列をparse_json関数を用いてVARIANT型に変換しています。

%sql
CREATE TABLE family_variant AS
SELECT family_name, PARSE_JSON(member) variant_column
FROM family_json


string型だったmemberカラムの中身が、キーと値のペアとしてvariant_columnカラムに格納されました。

続いて、JSONオブジェクトの中の値を取り出します。クエリ方法がstruct型とは異なり、「VARIANT列名:キー」を使用します。

%sql
select family_name, variant_column:name, variant_column:city
from family_variant

今回は検証していませんが、VARIANTデータ型を使用してDeltaテーブルに半構造化データを格納する際には、

  • バリアント列をGROUP BY句または ORDER BY句で使用できない
  • バリアント列に16 MiBよりを含めることが出来ない

等の制限が存在する (https://docs.databricks.com/aws/ja/delta/variant#limitations) ため、注意が必要です。

まとめ

今回は、JSONファイルからのデータの取り込みや、簡単な処理の方法について試した結果をご紹介しました。また、パブリックプレビュー段階の機能であるVARIANT型についても同様のデータを用いてデータの読み込みを行ってみました。
今後、より複雑な半構造化データの処理や正式リリース後のVARIANT型についても調査を行っていきたいと考えています。

以下の関連記事セクションに記載の記事でも、Spark初心者向けに、機能を使用してみた情報が紹介されています。気になった方は是非ご覧ください。

関連記事

https://zenn.dev/data_and_ai/articles/basic-operations-of-spark-rdd

https://zenn.dev/data_and_ai/articles/e61becccf64bea

https://zenn.dev/data_and_ai/articles/561cf22b09ca06

参考文献

Databricks公式ドキュメント

DXC Data&AI
DXC Data&AIPublication

Discussion