ノートブックを使用してデータを取り込む

Azure Databricks のノートブックは、さまざまなソースからデータを取り込むための柔軟なコード駆動型アプローチを提供します。 グラフィカル ツールが要件を満たしていない場合、またはデータ変換にカスタム ロジックが必要な場合は、ノートブックでインジェスト プロセスを完全に制御できます。 Python (PySpark)、SQL、Scala、または R を使用して、データ ソースへの接続、変換の適用、Unity カタログ テーブルへのデータの読み込みを行うことができます。

このユニットでは、バッチとストリーミングの両方のシナリオでノートブックを使用してデータを取り込む方法について説明します。

DataFrames を使用してバッチ データを取り込む

DataFrames は、Databricks でのバッチ データ インジェストの基礎です。 DataFrame は、リレーショナル データベース内のテーブルと同様に、名前付き列に編成されたデータの分散コレクションです。 spark.read API を使用して、さまざまなソースから DataFrame にデータを読み込みます。

次の例では、CSV ファイルを DataFrame に読み取ります。

df = (spark.read
    .format("csv")
    .option("header", "true")
    .option("inferSchema", "true")
    .load("/Volumes/catalog/schema/volume/data.csv"))

format()メソッドはファイルの種類を指定しますが、option()メソッドはデータの解析方法を構成します。 header オプションは、最初の行を列名として使用するように Spark に指示し、データ型inferSchema自動的に検出します。

さまざまなファイル形式から読み取る

Azure Databricks では、複数のファイル形式がサポートされています。 データ ソースに基づいて、形式とオプションを調整できます。

# Read JSON files
df_json = spark.read.format("json").load("/path/to/data.json")

# Read Parquet files (columnar format, efficient for analytics)
df_parquet = spark.read.format("parquet").load("/path/to/data.parquet")

# Read XML files (requires rowTag option)
df_xml = (spark.read
    .format("xml")
    .option("rowTag", "record")
    .load("/path/to/data.xml"))

<data>
  <record><id>1</id><name>Alice</name></record>
  <record><id>2</id><name>Bob</name></record>
</data>

Parquet ファイルには列形式でデータが格納されるため、分析ワークロードのクエリ パフォーマンスが向上します。 JSON ファイルは入れ子になった構造をサポートしており、API 応答で一般的です。 XML ファイルでは、各行にマップする要素を指定するために、 rowTag オプションが必要です。

外部データベースに接続する

リレーショナル データベースに格納されているデータの場合は、JDBC 接続を使用してソースから直接読み取ってください。

df_jdbc = (spark.read
    .format("jdbc")
    .option("url", "jdbc:sqlserver://server.database.windows.net:1433;database=mydb")
    .option("dbtable", "schema.tablename")
    .option("user", dbutils.secrets.get(scope="jdbc", key="username"))
    .option("password", dbutils.secrets.get(scope="jdbc", key="password"))
    .load())

資格情報をノートブックにハードコーディングするのではなく、Databricks シークレットに格納します。 dbtable オプションのサブクエリを使用して、クエリをデータベースにプッシュダウンすることもできます。

pushdown_query = "(SELECT id, name, amount FROM orders WHERE order_date > '2024-01-01') AS filtered_orders"
df = spark.read.format("jdbc").option("dbtable", pushdown_query).load()

Unity カタログ テーブルにデータを書き込む

データを読み込んで変換した後、Unity カタログ テーブルに書き込みます。 Azure Databricks では、ACID トランザクションとバージョン管理を提供する Delta Lake 形式が既定で使用されます。

df.write.mode("overwrite").saveAsTable("catalog.schema.table_name")

mode() オプションは、既存のデータの処理方法を制御します。

• overwrite: テーブルの内容を完全に置き換えます
• append: 既存のテーブルに行を追加します
• ignore: テーブルが既に存在する場合は何も行いません
• error: テーブルが存在する場合に例外をスローします (既定値)

構造化ストリーミングを使用してストリーミング データを取り込む

構造化ストリーミングを使用すると、受信したデータを継続的に処理できます。 spark.readの代わりに、spark.readStreamを使用して、新しいデータを段階的に処理するストリーミング DataFrame を作成します。

次の例では、Kafka からストリーミング データを読み取ります。

df_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "topic-name")
    .option("startingOffsets", "latest")
    .load())

startingOffsets オプションは、新しいクエリを開始するときに読み取りを開始する場所を決定します。 新しいデータにのみ latest を使用するか、トピックの先頭から読み取る earliest を使用します。 書き込みストリームでチェックポイントの場所を構成すると、後続の実行は最後に処理されたオフセットから自動的に再開され、 startingOffsetsは無視されます。

ファイル ベースのストリームを処理する

新しいファイルがディレクトリに到着すると、ファイル ソースからストリーミングすることもできます。

df_file_stream = (spark.readStream
    .format("csv")
    .option("header", "true")
    .schema(defined_schema)
    .load("/Volumes/catalog/schema/volume/incoming/"))

ストリーミング ファイル ソースの場合は、 inferSchemaを使用するのではなく、スキーマを明示的に定義する必要があります。 スキーマ推論は、データセット全体を読み取る必要があるため、ストリーミング ソースではサポートされていません。

ストリーミング データをテーブルに書き込む

writeStreamを使用して、ストリーミング データを Unity カタログ テーブルに保持します。

(df_stream
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("catalog.schema.streaming_table"))

出力モードによって、書き込まれるレコードが決まります。

• append: 新しい行のみが書き込まれます (ほとんどの操作では既定)。
• update: 変更された行が書き込まれます
• complete: すべての行が書き換えられます (集計で使用されます)

チェックポイントの場所には進行状況情報が格納され、中断された場合に停止した場所からストリームを再開できます。

半構造化データの処理

JSON などの半構造化データを使用する場合、VARIANT データ型は、入れ子になった構造を格納およびクエリするための柔軟性を提供します。

from pyspark.sql.functions import parse_json, variant_get, col

df = spark.read.text("/path/to/data.json")
df_variant = df.select(parse_json(col("value")).alias("data"))

その後、 variant_get 関数または SQL 構文を使用して、特定のフィールドに対してクエリを実行できます。

df_variant.select(variant_get(col("data"), "$.customer.name", "string")).display()

この方法は、JSON スキーマがレコード間で異なる場合や、時間の経過と同時に進化する場合に適しています。