PySpark チートシート: OVHcloud Data Platform 上の基本的および高度な関数#

#目的

このチートシートは、OVHcloud Data Platform 上のデータ処理のための基本的および高度な PySpark 関数の簡潔なガイドを提供します。これは、2025年1月のNYCイエローキャブ旅行記録（yellow_tripdata_2025_01.parquet、約3.5Mレコード）とタクシーゾーンルックアップテーブル（taxi_zone_lookup.csv、265レコード）を実践的な例として使用します。

中級ユーザーを対象としており、複雑な変換と分析のためのコア関数（filter、select、groupBy、join、udf）と高度な関数（window、pivot、approx_count_distinct、collect_list、explode、regexp_replace）をカバーしています。これらの例は実用的な応用を示しており、このチートシートはどのようなデータセットにも役立つ多才なリファレンスになります。

#必要条件

始める前に、以下を確認してください：

• データセット：（yellow_tripdata_2025_01.parquet、taxi_zone_lookup.csv）が、Connectors に利用可能で、Lakehouse Manager からアクセス可能であること。これらのデータセットは、公式のNYC TLC Trip Record Data ウェブサイトからダウンロードできます。
• ノートブック：PySpark を有効にした Jupyter ノートブック。

#セットアップ手順

1. Connectors：「NYC-taxi」というソースを作成し、ファイル（yellow_tripdata_2025_01.parquet、taxi_zone_lookup.csv）をアップロードし、そのスキーマを抽出します。
2. Lakehouse Manager：Lakehouse Manager に対応するテーブルを作成します。
3. DPE（Data Processing Engine）：これらのテーブル内にデータが読み込まれていることを確認します。
4. ノートブック：OVHcloud Data Platform 内で PySpark Jupyter ノートブックを開始します。

#PySpark 関数 チートシート

#ステップ 1: Spark の初期化とデータの読み込み

コードブロック

from forepaas.dwh import connect
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, unix_timestamp, hour, when, count, avg, udf, approx_count_distinct, collect_list, explode, regexp_replace, row_number
from pyspark.sql.types import FloatType
from pyspark.sql.window import Window
from forepaas.dwh.common import request as dwh_request, DwhRequestException
import io
import logging

# Setup logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Set up variables (DATASET, PROJECT_ID, YEAR, MONTH)
DATASET = "default_dataset"
PROJECT_ID = "PROJECT_ID" # Make sure to replace this with your actual PROJECT_ID
YEAR = "2025"
MONTH = "01"

# Initialize SparkSession
spark = SparkSession.builder.appName("PySpark_Advanced_Cheat_Sheet").getOrCreate()
logging.info(f"Spark Version: {spark.version}")

# Connect to Lakehouse
cn_prim = connect("dwh/default_dataset/")

# Load data
taxi_df = cn_prim.query(f"SELECT * FROM db_{PROJECT_ID}_{DATASET}.{DATASET}.yellow_tripdata_{YEAR}_{MONTH}")
zones_df = cn_prim.query(f"SELECT LocationID, Borough, Zone FROM db_{PROJECT_ID}_{DATASET}.{DATASET}.taxi_zone_lookup")
taxi_df.cache()
zones_df.cache()
logging.info(f"Taxi Records: {taxi_df.count()}")
logging.info(f"Zones Records: {zones_df.count()}")

このコードブロックで使用される関数:

• SparkSession.builder.appName().getOrCreate():

  • 何: PySpark セッションと DataFrame API を初期化します。これは Spark 機能を使用するためのエントリーポイントです。
  • 利点: 分散データ処理のための環境を設定します。

• cn_prim = connect("dwh/default_dataset/"):

  • 何: forepaas.dwh ライブラリを使用して、指定された Lakehouse データセットに接続します。
  • 利点: Lakehouse に保存されたテーブルをクエリや操作することを可能にします。

• cn_prim.query(sql_query):

  • 何: cn_prim 接続を介してアクセス可能なテーブルに対して SQL クエリを実行します。データを Spark DataFrame として取得します。
  • 利点: Connectors と Lakehouse から Spark DataFrame にデータを読み込むための簡単な方法を提供します。

• DataFrame.cache():

  • 何: DataFrame を最初に計算したときにメモリにキャッシュするようにマークします。この DataFrame に対する後のアクションはキャッシュから読み込みます。
  • 利点: 同じ DataFrame に対する操作を大幅に高速化します。特に、大規模なデータセット（約 3.5M レコード）に対して反復アルゴリズムや複数の変換を行う場合に有用です。

• DataFrame.count():

  • 何: 計算をトリガーし、DataFrame の行の総数を返します。
  • 利点: データが読み込まれたことを迅速に確認し、レコード数を確認するために使用されます。

このコードブロックの出力:

• Spark バージョン（例: Spark Version: 3.4.1）
• Taxi Records: ~3,500,000（実際のカウントは若干異なる場合があります）
• Zones Records: 265

#ステップ 2: データのクリーンアップと変換

コードブロック

# Clean and transform
cleaned_df = taxi_df \
    .filter(
        (col("tpep_pickup_datetime").isNotNull()) &
        (col("fare_amount") > 0) &
        (col("trip_distance") > 0)
    ) \
    .select(
        "tpep_pickup_datetime",
        "tpep_dropoff_datetime",
        "trip_distance",
        "fare_amount",
        col("pulocationid").cast("double").alias("pulocationid"),
        regexp_replace(col("store_and_fwd_flag"), "^[Yy]$", "Yes").alias("store_and_fwd_flag")
    ) \
    .withColumn(
        "trip_duration",
        unix_timestamp("tpep_dropoff_datetime") - unix_timestamp("tpep_pickup_datetime")
    ) \
    .withColumn(
        "pickup_hour",
        hour("tpep_pickup_datetime")
    ) \
    .withColumn(
        "trip_duration",
        when(col("trip_duration") > 3600, 3600).otherwise(col("trip_duration"))
    ) \
    .filter(col("trip_duration") >= 60)

logging.info(f"Cleaned Records: {cleaned_df.count()}")

このコードブロックで使用される関数:

• DataFrame.filter(condition):

  • 何: 指定された条件に基づいて DataFrame の行をフィルタリングし、条件を満たす行のみを含む新しい DataFrame を返します。
  • 利点: データのクリーンアップに不可欠で、無効または関連性のないレコードを削除します。

• pyspark.sql.functions.col(column_name):

  • 何: DataFrame の列を参照します。これにより、列に対して様々な変換と操作を適用できます。
  • 利点: DataFrame 列を使用した式を構築する方法を提供します。

• DataFrame.select(columns):

  • 何: 一連の式（列または列ベースの変換）をプロジェクトし、選択した列のみを含む新しい DataFrame を返します。
  • 利点: 列のサブセット化に便利で、関連するフィールドのみを選択することで DataFrame のサイズを削減します。

• Column.cast(dataType):

  • 何: 列のデータ型を指定された dataType に変換します。
  • 利点: 計算やダウンストリームのプロセスでのデータ型の互換性を確保します（例: pulocationid を double にキャスト）。

• pyspark.sql.functions.regexp_replace(column, pattern, replacement):

  • 何: 列のテキストデータ内の文字列パターンのすべての出現を指定された置換文字列に置き換えます。
  • 利点: データの標準化と文字列フィールドのクリーンアップに優れています（例: "Y" を "Yes" に変換）。

• DataFrame.withColumn(colName, col):

  • 何: 指定された式に基づいて新しい列を追加したり、既存の列を置き換えたりして新しい DataFrame を返します。
  • 利点: フィーチャーエンジニアリングと新しい派生列のオンデマンド作成に不可欠です。

• pyspark.sql.functions.unix_timestamp(timestamp_column):

  • 何: タイムスタンプ文字列またはタイムスタンプ列を Unix タイムスタンプ（1970-01-01 00:00:00 UTC 以降の秒数）に変換します。
  • 利点: タイムスタンプの差を計算するなどの数値的な時間計算を容易にします。

• pyspark.sql.functions.hour(timestamp_column):

  • 何: タイムスタンプ列から時間成分を抽出します。
  • 利点: 時系列分析に便利で、データ内の時間帯ごとのパターンやトレンドを特定できます。

• pyspark.sql.functions.when(condition, value).otherwise(other_value):

  • 何: 条件付きロジックを実装します。condition が true の場合、列は value を取得します。それ以外の場合は other_value を取得します。チェーン化できます。
  • 利点: アウトライアの処理、ビジネスルールの適用、または特定の条件に基づいてデータを分類するのに効果的です（例: trip_duration のキャップ）。

このコードブロックの出力:

• Cleaned Records: ~2,700,000 – ~2,800,000（データ品質に基づいて実際のカウントは異なる場合があります）。

#ステップ 3: UDF（ユーザー定義関数）によるカスタムロジック

コードブロック

# UDF for fare efficiency (fare per minute)
def fare_efficiency(fare, duration):
    return fare / (duration / 60) if duration > 0 else 0.0

fare_efficiency_udf = udf(fare_efficiency, FloatType())

# Apply UDF
transformed_df = cleaned_df \
    .withColumn("fare_efficiency", fare_efficiency_udf(col("fare_amount"), col("trip_duration")))

transformed_df.show(5)

このコードブロックで使用される関数:

• udf(func, returnType):

  • 何: Python 関数を PySpark のユーザー定義関数（UDF）として登録します。これにより、カスタム Python ロジックを Spark DataFrame 列に適用できます。
  • 利点: ネイティブ PySpark 関数に利用できない複雑な計算をカスタマイズできます（例: カスタムロジックを使用して「1分あたりの料金」を計算）。

• FloatType()（pyspark.sql.types から）：

  • 何: UDF の戻り値のデータ型を Float（単精度浮動小数点数）として指定します。
  • 利点: カスタム関数の出力が Spark DataFrame で正しく型付けされることを確保します。

• DataFrame.withColumn(colName, col):

  • 何: （ステップ 2 から再利用）式の結果に基づいて新しい列を追加したり、既存の列を置き換えたりします。
  • 利点: ここで、新しく定義された fare_efficiency_udf を適用して fare_efficiency 列を作成します。

このコードブロックの出力:

• DataFrame.show(5) 出力、transformed_df の最初の 5 行を表示し、新しく追加された fare_efficiency 列を含みます。
  • 例: fare_amount が 15.0 で trip_duration が 600 秒（10 分）の場合、fare_efficiency は 1.5 ($/min) となります。

#ステップ 4: 高度なウィンドウ関数とジョイン

コードブロック

# Define window for ranking trips by fare within borough
window_spec = Window.partitionBy("pickup_borough").orderBy(col("fare_amount").desc())

# Join with zones and rank trips
joined_df = transformed_df \
    .join(
        zones_df,
        transformed_df.pulocationid == zones_df.LocationID,
        "left"
    ) \
    .withColumnRenamed("Borough", "pickup_borough") \
    .drop("LocationID") \
    .filter(col("pickup_borough").isNotNull()) \
    .withColumn("fare_rank", row_number().over(window_spec))

logging.info(f"Joined Records: {joined_df.count()}")
joined_df.filter(col("fare_rank") <= 3).show()

このコードブロックで使用される関数:

• Window.partitionBy(*cols).orderBy(*cols):

  • 何: ウィンドウ仕様を定義します。partitionBy は行をグループに分け、orderBy は各パーティション内の行の論理的な順序を定義します。
  • 利点: ランキング、リード/ラグ、累積合計などの高度な分析操作を有効にするために不可欠です。これらは定義された行のサブセットで動作します。

• pyspark.sql.functions.row_number():

  • 何: ウィンドウ関数で、ウィンドウ仕様で定義された順序に基づいて、各パーティション内の各行に一意の連続番号を割り当てます。
  • 利点: レコードのランキングに最適です（例: メトリクスに基づいてトップ N レコードを特定する、トップファーレート）。

• DataFrame.join(other_df, on=None, how=None):

  • 何: 指定されたジョイン条件（on）とジョインタイプ（how、例: "inner"、"left"、"right"）に基づいて 2 つの DataFrame を結合します。
  • 利点: 異なるソースからの関連情報を統合することでデータを豊かにします（例: タクシー乗車データをゾーンルックアップデータと結合）。

• DataFrame.withColumnRenamed(existing, new):

  • 何: 既存の列を名前を変更した新しい DataFrame を返します。
  • 利点: スキーマを明確にし、読みやすさを向上させます。特にジョイン後に列名が曖昧になる場合に便利です。

• DataFrame.drop(*cols):

  • 何: 指定された列を削除した新しい DataFrame を返します。
  • 利点: 不要な列を削除することで DataFrame のサイズと複雑さを管理し、メモリを節約します。

このコードブロックの出力:

• Joined Records: ~2,600,000 – ~2,700,000（実際のカウントは異なる場合があります）。
• DataFrame.show() 出力、fare_rank が 3 以下の行を表示し、各区のトップ 3 ファーレートを表示します。

#ステップ 5: データの集約とピボット

コードブロック

# Aggregate: unique zones and trips per borough
agg_df = joined_df \
    .groupBy("pickup_borough") \
    .agg(
        approx_count_distinct("pulocationid").alias("unique_zones"),
        count("*").alias("num_trips")
    )

# Pivot: avg fare by hour and borough
pivot_df = joined_df \
    .groupBy("pickup_hour") \
    .pivot("pickup_borough") \
    .agg(avg("fare_amount")) \
    .orderBy("pickup_hour")

agg_df.show()
pivot_df.show()

このコードブロックで使用される関数:

• DataFrame.groupBy(*cols):

  • 何: DataFrame を 1 つ以上の指定された列でグループ化し、集約計算の準備をします。
  • 利点: 異なるカテゴリや次元に基づいてデータの要約と分析を可能にします。

• DataFrame.agg(*exprs):

  • 何: グループ化されたデータに集約関数を適用し、要約統計を計算します。
  • 利点: 各グループのカウント、合計、平均などのメトリクスを計算するために使用されます。

• pyspark.sql.functions.approx_count_distinct(column):

  • 何: グループ内の一意の項目の近似カウントを返します。HyperLogLog++ アルゴリズムを使用します。
  • 利点: 非常に大規模なデータセットでは、countDistinct よりも大幅に高速でメモリ効率が良く、正確なカウントが必ずしも必要ない場合に便利です。

• pyspark.sql.functions.count(column):

  • 何: 列の非 NULL 値の数をカウントするか、count("*") を使用してグループ内のすべての行をカウントします。
  • 利点: 各集約グループのサイズまたは特定の値の発生回数を集計します。

• DataFrame.pivot(pivot_column):

  • 何: 指定された列の一意の値を新しい列に変換（ピボット）して DataFrame を回転させます。その後の集約が必要です。
  • 利点: 通常はレポートや横断的分析に適した広いテーブルを作成し、カテゴリ間の値を直接比較できます。

• pyspark.sql.functions.avg(column):

  • 何: 数値列の平均値を計算します。
  • 利点: 各グループ内の定量データの中央値を提供します。

• DataFrame.orderBy(*cols, ascending=True):

  • 何: 1 つ以上の列に基づいて DataFrame の行を昇順または降順で並べ替えます。
  • 利点: 出力を整理し、データを論理的な順序で提示します。

このコードブロックの出力:

• agg_df.show(): pickup_borough、unique_zones（近似）および num_trips を表示するテーブル（例: マンハッタンには約 60 の一意のゾーンと約 2M の乗車があります）。
• pivot_df.show(): 行として pickup_hour を、列として pickup_borough を表示するピボットテーブル、各セルに平均 fare_amount が含まれます。

#ステップ 6: リストの収集と展開

コードブロック

# Collect zones per borough
list_df = joined_df \
    .groupBy("pickup_borough") \
    .agg(collect_list("Zone").alias("zones_list"))

# Explode zones list
exploded_df = list_df \
    .select("pickup_borough", explode(col("zones_list")).alias("zone"))

exploded_df.show(10)

このコードブロックで使用される関数:

• pyspark.sql.functions.collect_list(column):

  • 何: 指定された列内の各グループの非 NULL 値をすべて Python リストに集約する集約関数です。
  • 利点: 各要素が元のグループのレコードに対応する配列のような構造を作成するのに便利です。

• pyspark.sql.functions.explode(array_column):

  • 何: 配列（リスト）またはマップを含む列を、配列/マップ内の各要素に対応する個別の行に変換します。配列に N 個の要素がある場合、その元の行に対して N 個の行を作成します。
  • 利点: ネストされたデータ構造をフラット化し、個々の要素を個別のレコードとして処理または表示できます。

このコードブロックの出力:

• DataFrame.show(10) 出力、pickup_borough と exploded zone 列を含むテーブルを表示し、各区の各一意のゾーンが個別の行を取得します（例: マンハッタンの場合、「マンハッタン | ミッドタウン」、「マンハッタン | アッパーイーストサイド」などの複数の行が表示されます）。

#ステップ 7: DataFrame をバケットに保存してテーブルを作成（高度）

コードブロック

def create_table_from_this_dataframe(dataframe, dataset, table_name, bucket, source_bucket):
    logging.info(f"We will create a source (bucket) - {source_bucket} where we will store the new table - {table_name} - and automatically load it")

    # Creating bucket to store the table
    cn_datastore = connect('data_store')
    logging.info(f"{cn_datastore.list()} - Before creating new bucket")
    cn_datastore.create_bucket(bucket)
    logging.info(f"{cn_datastore.list()} - After adding new bucket")
    cn_bucket = connect('data_store/' + bucket)

    get_dbs = dwh_request(f"v4/databases", method="GET")
    data = get_dbs.json()
    db_exist = next((item['_id'] for item in data if item.get('display_name') == source_bucket and item.get("package") == "data-store"), None)

    if db_exist is None:
        # Creating source where we will use the new bucket created to get access to the table
        new_source = {"type":"protocol","package":"data-store","parameters":{"path":"","bucket":bucket},"default":False,"level":"source","display_name":source_bucket}
        new_source_bucket = dwh_request(f"v4/databases", method="POST", json=new_source)
        logging.info(f"New source added with the bucket: {bucket} - source name: {source_bucket}")
    else:
        logging.info("Source already exist")

    # Call to API - To get dataset id
    get_database_id = dwh_request(f"v4/databases", method="GET")
    database_all = get_database_id.json()
    # Filter to get the corresponding _id for the database
    dataset_id = next((item['_id'] for item in database_all if item.get('name') == dataset), None)

    logging.info(f"dataset_id : {dataset_id}")

    # Convert table to Pandas and serialize to CSV in BytesIO
    try:
        # Convert to Pandas DataFrame
        table = dataframe.toPandas()

        # Create BytesIO buffer and write CSV
        data = io.BytesIO()
        table.to_csv(data, index=False, encoding='utf-8')
        data.seek(0)  # Reset buffer position

        # Upload to bucket
        file_path = f"{table_name}.csv"
        etag = cn_bucket.put(file_path, data, data.getbuffer().nbytes)
        logging.info(f"DataFrame uploaded to bucket {bucket}/{file_path} with ETag: {etag}")

        # Verify bucket contents
        files = cn_bucket.list()
        logging.info(f"Bucket contents: {files}")
    except Exception as e:
        logging.error(f"Failed to save to bucket: {e}")
        raise

    # Call to API - To add the file into the source
    table_config = {"display_name":file_path,"progress":None,"physical_status":None,"parameters":{},"filename":file_path}
    res_table = dwh_request(f"v4/databases/{source_bucket}/tables/{file_path}", method="PUT", json=table_config)

    # Call to API - To get the corresponding ID for the file_path added in the source
    get_template_catalog_object = dwh_request(f"v4/tables", method="GET")
    data = get_template_catalog_object.json()
    file_path_source_id = next((item['_id'] for item in data if item.get('filename') == file_path), None)

    logging.info(f"file_path_source_id : {file_path_source_id}")

    # In case the table already exist and you made some modification on it
    auto_build_table_DELETE = dwh_request(f"v4/logical/objects/{table_name}", method="DELETE") 

    # Call to API - To launch the build of the table on the corresponding dataset and load the specific data
    config_build = {"database_id":dataset_id,"display_name":table_name,"name":table_name,"type":"prim","load_data":True,"build_table":True,"templated_from":"data_catalog","template_catalog_object":file_path_source_id}
    auto_build_table = dwh_request(f"v4/logical/objects", method="POST", json=config_build)

    logging.info(f"You can check the build of the table {table_name} on the Lakehouse Manager screen")

# Use the function create_table_from_this_dataframe:
create_table_from_this_dataframe(pivot_df,"default_dataset","taxi_pivot_table","new_bucket","new_source_bucket")

このコードブロックで使用される関数（および create_table_from_this_dataframe 内）:

• DataFrame.toPandas():

  • 何: Spark DataFrame を Pandas DataFrame に変換します。これにより、すべての分散データがドライバーノードに収集されます。
  • 利点: Pandas 特有の関数を使用してローカルデータ操作とファイルシリアライゼーション（例: to_csv）を可能にします。
  • 警告: 非常に大規模なデータセットでは、ドライバでメモリ不足エラーが発生する可能性があるため、注意して使用してください。

• Pandas_DataFrame.to_csv(path_or_buffer, index=False, encoding='utf-8'):

  • 何: Pandas DataFrame をカンマ区切り値（CSV）ファイルに書き込みます。
  • 利点: データを保存と取得に適した標準テキスト形式にシリアライズします。

• io.BytesIO():

  • 何: Python の io モジュールからのクラスで、メモリ内のバイナリストリームを作成します。これはファイルオブジェクトのように動作します。
  • 利点: 物理ファイルとやり取りしているかのようにバイトを書き込みおよび読み取ることができ、ディスクに保存せずにサービスに直接データを転送するのに便利です。

• cn_datastore.create_bucket(bucket_name)（forepaas.dwh から）：

  • 何: OVHcloud Data Platform のデータストア内に新しいストレージバケットを作成します。
  • 利点: ファイル、中間処理データ、最終処理データを保存するための専用の場所を提供します。

• cn_bucket.put(file_path, data, size)（forepaas.dwh から）：

  • 何: 通常は BytesIO バッファから、接続されたバケット内の指定されたパスにデータをアップロードします。
  • 利点: 処理済みデータ（例: pivot_df からの CSV）を OVHcloud クラウドストレージに永続化します。

• cn_bucket.list()（forepaas.dwh から）：

  • 何: 接続されたバケット内のファイルとサブディレクトリのリストを取得します。
  • 利点: ファイルがバケットに正常にアップロードされたことを確認するために使用されます。

• dwh_request(path, method, json)（forepaas.dwh.common から）：

  • 何: OVHcloud Data Platform のバックエンドサービスに直接 HTTP API 呼び出しを行うユーティリティ関数です。
  • 利点: データソースの作成、Connectors にファイルの登録、Lakehouse Manager でテーブルのビルドをトリガーするなどのタスクを自動化するための細かい制御を提供します。これらは常に connect オブジェクトを介して公開されるわけではありません。

このステップのプロセスの説明:

この高度なステップでは、処理済み Spark DataFrame（pivot_df）を OVHcloud Data Platform の Lakehouse に新しいテーブルとして永続化する方法を紹介します。PySpark、Pandas、および直接の API 呼び出しの組み合わせを活用します:

1. バケット作成: OVHcloud データストアに新しいバケット（new_bucket）が既に存在しない場合に作成されます。
2. ソース作成: 新しいデータソース（new_source_bucket）が構成され、このバケットにリンクされ、Connectors がその中のファイルを発見できるようになります。
3. データシリアライゼーションとアップロード: pivot_df は Pandas DataFrame に変換され、メモリ内バッファ（io.BytesIO）内で CSV 形式にシリアライズされます。この CSV データは新しく作成されたバケットにアップロードされます。
4. テーブル登録とビルド: 直接の API 呼び出し（dwh_request）を介して、アップロードされた CSV ファイルが Connectors で「物理テーブル」として登録され、Lakehouse Manager で「論理テーブル」（taxi_pivot_table）が作成され、登録されたソースからデータがロードされます。

#ステップ 8: テーブルの存在を確認

コードブロック

# Verify table exists in Lakehouse
try:
    test_result = cn_prim.query(f"SELECT * FROM db_{PROJECT_ID}_{DATASET}.{DATASET}.taxi_pivot_table")
    logging.info(f"taxi_pivot_table Records: {test_result.count()}")
except Exception as e:
    logging.error(f"Failed to query table - please ensure it's the correct table")

このコードブロックで使用される関数:

• cn_prim.query(sql_query):

  • 何: （ステップ 1 から再利用）確立された cn_prim 接続を介して Lakehouse のテーブルに対して SQL クエリを実行し、結果を Spark DataFrame として取得します。
  • 利点: ここで、taxi_pivot_table が正常に作成され、Lakehouse でクエリ可能であることを確認するために使用されます。

• DataFrame.count():

  • 何: （ステップ 1 から再利用）クエリされた DataFrame の行の総数を返します。
  • 利点: 新しく作成されたテーブルに期待されるデータとレコードが含まれていることを確認します。

このコードブロックの出力:

• 成功した場合: taxi_pivot_table Records: [Number of rows in pivot_df]（例: 24 時間が存在する場合は taxi_pivot_table Records: 24）。
• 失敗した場合: テーブルをクエリできないことを示す logging.error からのエラーメッセージ。

#ステップ 9: クリーンアップと Spark セッションの停止

コードブロック

spark.catalog.clearCache()
spark.stop()
logging.info("Spark cache cleared and Spark session stopped.")

このコードブロックで使用される関数:

• SparkSession.catalog.clearCache():

  • 何: Spark の内部キャッシュをクリアし、キャッシュされた DataFrame と RDD が占有するメモリを解放します。
  • 利点: 複雑な操作の後やキャッシュされたデータが必要なくなった場合にリソースを解放するために不可欠です。長時間実行されるアプリケーションやインタラクティブセッションでメモリ不足を防ぐのに役立ちます。

• SparkSession.stop():

  • 何: SparkSession を終了し、SparkContext と関連するすべてのリソースを解放します。
  • 利点: Spark アプリケーションのクリーンシャットダウンを確保し、クラスターリソースを解放します。Spark アプリケーションの終了時にこの関数を呼び出すことが重要で、リソースリークを防ぎます。

このコードブロックの出力:

• INFO - Spark cache cleared and Spark session stopped.
  • このメッセージは、キャッシュがクリアされ、Spark セッションが正常に終了したことを確認します。通常、カーネルがアイドル状態であるかシャットダウンしたことを示すノートブック環境からのさらに多くの出力を確認できます。

#結論と次のステップ

このPySparkチートシートでは、OVHcloud Data Platform上でのデータ処理に必要な基本的および高度な関数のハンズオン概要を提供しました。Sparkの初期化方法、データの読み込みとクリーンアップ、UDFを使用したカスタムロジックの適用、ウィンドウ関数とピボットを使用した複雑な集計の実行、そしてLakehouseへのデータ永続化の管理方法を学びました。

このガイドは、PySpark開発を加速するための実用的なリファレンスとして役立ちます。これらのスキルを現実のシナリオに適用し続けるために、以下のアクションをお勧めします。

• チュートリアルを探索する: NYC Taxi Dataset Analysisチュートリアルに深く掘り下げ、エンドツーエンドのデータパイプラインと機械学習モデルを構築してください。
• 知識を深める: すべての関数について詳細な情報を得るために、公式PySparkドキュメントを参照してください。

コーディングとデータ分析を楽しんでください！ 🚀

#さらに深く学ぶ

トレーニングや技術サポートが必要な場合は、営業担当者にお問い合わせください、またはこのリンクをクリックして見積もりを依頼し、プロフェッショナルサービスの専門家にプロジェクトのカスタム分析を依頼してください。

Data Platformの開発チームと直接質問し、フィードバックを共有し、交流するには、専用のDiscordチャネルをご利用ください。

OVHcloudサービスについてサポートが必要な場合は、ヘルプセンターでリクエストを作成してください。

ユーザーコミュニティに参加してください。