Sparkでは、デフォルトの結合タイプは内部結合で、両方のDataFrameで一致するキーを持つ行のみを返します。左側のDataFrame（purch_df）のすべてのレコードを保持し、右側のDataFrame（cust_df）の一致するレコードを含めるには、左外部結合を使用する必要があります。
結合タイプを「左」に指定することで、修正されたコードではpurch_dfのすべてのレコードが保持され、cust_dfの一致するレコードも含まれるようになります。purch_dfのレコードのうち、cust_dfに対応するレコードがない場合、cust_dfの列にはnull値が設定されます。
このアプローチは標準の SQL 結合操作と一致しており、PySpark の DataFrame API でサポートされています。

Spark (以前の Koalas) 上の Pandas API では、メソッド .to_spark() が pyspark.pandas.DataFrame を PySpark DataFrame に変換します。
正しい使い方:
spark_df = pdf.to_spark()
これにより、Spark 上の Pandas API と PySpark SQL API 間の相互運用性が実現し、開発者は変換やパフォーマンスの最適化のために両者をシームレスに切り替えることができます。
他のオプションが間違っている理由:
A (to_pandas): PySpark DataFrame ではなく、ローカルの Pandas DataFrame に変換します。
C (to_dataframe): 有効な API メソッドではありません。
D (spark): 既存の DataFrame メソッドではありません。
参照：
PySpark Pandas API リファレンス - DataFrame.to_spark() メソッド。
Databricks 試験ガイド (2025 年 6 月) : セクション「Apache Spark での Pandas API の使用」 - DataFrame の変換と相互運用性について説明します。

データセットのパーティション分割では、パーティション列の値に基づいてデータが個別のディレクトリに分割されます。クエリがパーティション列でフィルタリングされると、Spark は不要なパーティションを除外します。つまり、フィルタ条件に一致するファイルのみを読み取ります。
アドバンテージ：
関連するデータのサブセットのみをスキャンすることで I/O を削減し、パフォーマンスを向上させます。
例：
/データ/売上/年=2023/月=10/...
/データ/売上/年=2024/月=01/...
WHERE year = 2024 をフィルタリングするクエリは、関連するパーティションのみを読み取ります。
他のオプションが間違っている理由:
A: 圧縮はパーティション分割とは無関係です。
B: 手動で管理されない限り、Spark はパーティションを自動的にクリーンアップしません。
C: パーティショニングによって Spark がデータ全体をメモリにロードするわけではありません。
参照：
Databricks 試験ガイド (2025 年 6 月) : セクション「Spark SQL の使用」 - 最適化されたデータ取得のためのパーティション分割とプルーニング。
Spark SQL ドキュメント - DataFrameWriter の partitionBy() とクエリの最適化。

正解は D です。これは、適切な時間ベースのウィンドウ集計とウォーターマークを使用しているためです。これは、イベント時間データの時間ベースの集計に Spark Structured Streaming で必要なパターンです。
構造化ストリーミングに関する Spark 3.5 ドキュメントより:
イベントタイム列にスライディングウィンドウを定義し、groupByとwindow()を使用してそれらのウィンドウの集計を計算できます。遅延データを処理するには、withWatermark()を使用して、遅延データの到着許容範囲を指定します。(出典: Structured Streaming Programming Guide) オプションDでは、以下の使用法が採用されています。
パイソン
コピー編集
.groupBy("sensor_id", window("timestamp", "5分"))
.agg(avg("温度").alias("avg_temp"))
各sensor_idについて、5分間のイベント時間ウィンドウにわたって平均温度が計算されることを保証します。ロジックを完成させるために、パイプラインの早い段階でwithWatermark("timestamp", "5 minutes")を使用して、遅延イベントを処理することを前提としています。
他のオプションが間違っている理由の説明:
オプション A は、静的 DataFrame またはバッチ クエリに適用される Window.partitionBy を使用しますが、ストリーミング集計には適していません。
オプション B では時間ウィンドウが適用されないため、5 分間の移動平均は計算されません。
オプション C では、集計後に withWatermark() が誤って適用され、時間ウィンドウが含まれていないため、必要な時間ベースのグループ化が欠落しています。
したがって、オプション D は、時間ウィンドウ ストリーミング集約を計算するためのすべての要件を満たす唯一のオプションです。

シャッフルは、groupBy、reduceByKey、joinなどの操作で発生し、これらの操作ではデータがパーティション間で移動されます。repartition()操作でもシャッフルが発生する可能性がありますが、このコンテキストでは集計の後に発生します。
オプション D では、groupBy の後に agg を実行すると、ノード間のグループ化によりシャッフルが発生します。
再パーティション(10)はパーティション変換ですが、データがすでにグループ化されているため、新たなシャッフルは行われません。
このシーケンス (シャッフル (groupBy)、その後に非シャッフル (repartition)) は正しいです。
オプション A はその逆を行います。フィルターはシャッフルを引き起こしませんが、groupBy はシャッフルを引き起こします。これにより順序が間違ってしまいます。