Spark を使用してデータ ファイルを操作する

ノートブックを設定してクラスターにアタッチした後、Spark を使用してデータ ファイルの読み取りと処理を行うことができます。 Spark では、CSV、JSON、Parquet、ORC、Avro、Delta などのさまざまな形式がサポートされており、Databricks には、ワークスペース、Azure Data Lake または Blob Storage、またはその他の外部システムに格納されているファイルにアクセスするための組み込みのコネクタが用意されています。

ワークフローは通常、次の 3 つの手順に従います。

1. 正しい形式とパスを使用して spark.read を使用して Spark DataFrame にファイルを読み取ります。 CSV や JSON などの生のテキスト形式を読み取る場合、Spark はスキーマ (列名とデータ型) を推測できますが、低速または信頼性の低い場合があります。 運用環境では、データが一貫して効率的に読み込まれるようにスキーマを明示的に定義することをお勧めします。

2. SQL 操作または DataFrame 操作 (行のフィルター処理、列の選択、値の集計など) を使用して、DataFrame を探索して変換します。

3. 選択した形式で結果をストレージに書き戻します。

Spark でのファイルの操作は、小規模および大規模なデータセット間で一貫するように設計されています。 小規模な CSV ファイルのテストに使用されるのと同じコードは、Spark がクラスター全体に作業を分散するため、はるかに大きなデータセットでも機能します。 これにより、迅速な探索からより複雑なデータ処理へのスケールアップが容易になります。

データフレームにデータを読み込む

データフレームを使用してデータを操作する方法を確認するには、仮定の例を考えてみましょう。 Unity カタログ ボリュームに格納 products.csvという名前 のコンマ区切りのテキスト ファイルに次のデータがあるとします。

ProductID,ProductName,Category,ListPrice
771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

Spark ノートブックでは、次の PySpark コードを使用してデータフレームにデータを読み込み、最初の 10 行を表示できます。

%python
df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/products.csv',
    format='csv',
    header=True,
    inferSchema=True
)
display(df.limit(10))

最初の%python行はマジックと呼ばれ、このセルで使用されている言語が Python であることを Spark に伝えます。 製品データの例と同等の Scala コードを次に示します。

%scala
val df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("/Volumes/my_catalog/my_schema/my_volume/products.csv")
display(df.limit(10))

マジック %scala は Scala を指定するために使用されます。

前に示した両方の例では、次のような出力が生成されます。

データフレーム スキーマを指定する

前の例では、CSV ファイルの最初の行に列名が含まれており、コードによって header オプションと inferSchema オプションが設定されているため、Spark は含まれるデータから各列のデータ型を推論しました。 また、データの明示的なスキーマを指定することもできます。これは、次の CSV の例のように、データ ファイルに列名が含まれていない場合に便利です。

771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

次の PySpark の例は、 product-data.csvという名前 のファイルから読み込まれるデータフレームのスキーマをこの形式で指定する方法を示しています。

from pyspark.sql.types import *
from pyspark.sql.functions import *

productSchema = StructType([
    StructField("ProductID", IntegerType()),
    StructField("ProductName", StringType()),
    StructField("Category", StringType()),
    StructField("ListPrice", FloatType())
    ])

df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/product-data.csv',
    format='csv',
    schema=productSchema,
    header=False)
display(df.limit(10))

ここでも、結果は次のようになります。

データフレームのフィルター処理とグループ化を行う

Dataframe クラスのメソッドを使用して、含まれているデータをフィルター処理、並べ替え、グループ化、操作できます。 たとえば、次のコード例では、select メソッドを使用して、前の例の製品データを含む df データフレームから ProductID 列と ListPrice 列を取得します。

pricelist_df = df.select("ProductID", "ListPrice")

このコード例の結果は次のようになります。

ほとんどのデータ操作メソッドと共通して、 select は新しいデータフレーム オブジェクトを返します。

メソッドを "チェーン" して、変換されたデータフレームを作成する一連の操作を実行できます。 たとえば、次のコード例では、selectメソッドと where メソッドを連結して、Mountain Bikes または Road Bikes のカテゴリを持つ製品の ProductName 列と ListPrice 列を含む新しいデータフレームを作成します。

bikes_df = df.select("ProductName", "ListPrice").where((df["Category"]=="Mountain Bikes") | (df["Category"]=="Road Bikes"))
display(bikes_df)

このコード例の結果は次のようになります。

データをグループ化して集計するには、 groupBy メソッドと集計関数を使用できます。 たとえば、次の PySpark コードでは、各カテゴリの製品数をカウントします。

counts_df = df.select("ProductID", "Category").groupBy("Category").count()
display(counts_df)

このコード例の結果は次のようになります。

Spark で SQL 式を使用する

Dataframe API は Spark SQL という名前の Spark ライブラリの一部であり、データ アナリストは SQL 式を使用してデータのクエリと操作を行います。

Spark カタログでデータベース オブジェクトを作成する

Spark カタログは、ビューやテーブルなどのリレーショナル データ オブジェクトのメタストアです。 Spark ランタイムは、カタログを使用して、Spark でサポートされている任意の言語で記述されたコードを、一部のデータ アナリストや開発者にとってより自然な SQL 式と統合できます。

Spark カタログでクエリを実行するためにデータフレーム内のデータを使用できるようにする最も簡単な方法の 1 つは、次のコード例に示すように、一時ビューを作成することです。

df.createOrReplaceTempView("products")

"ビュー" は一時的なもので、現在のセッションの終了時に自動的に削除されます。 また、カタログに保持される "テーブル" を作成して、Spark SQL を使用してクエリを実行できるデータベースを定義することもできます。

Spark SQL API を使用してデータのクエリを実行する

任意の言語で記述されたコードで Spark SQL API を使用して、カタログ内のデータに対してクエリを実行できます。 たとえば、次の PySpark コードでは、SQL クエリを使用して 製品 ビューからデータフレームとしてデータを返します。

bikes_df = spark.sql("SELECT ProductID, ProductName, ListPrice \
                      FROM products \
                      WHERE Category IN ('Mountain Bikes', 'Road Bikes')")
display(bikes_df)

このコード例の結果は、次の表のようになります。

SQL コードを使用する

前の例では、Spark SQL API を使用して、Spark コードに SQL 式を埋め込む方法を示しました。 また、ノートブックで %sql マジックを使用して、次のようにカタログ内のオブジェクトに対してクエリを行う SQL コードを実行することもできます。

%sql

SELECT Category, COUNT(ProductID) AS ProductCount
FROM products
GROUP BY Category
ORDER BY Category

この SQL コード例では、次のように、ノートブックにテーブルとして自動的に表示される結果セットが返されます。