各トリガーで完全な更新結果を出力する必要があるストリーミング集計の正しい出力モードは「complete」です。
公式ドキュメントより:
「完全: トリガーが発生するたびに、更新された結果テーブル全体がシンクに出力されます。」これは、キーでグループ化されたカウントや平均など、結果テーブルが時間の経過とともに増分的に変化する集計に最適です。
append: 新しく追加された行のみを出力します
置換と集計: 出力モードに無効な値

Spark の partitionBy() メソッドは、指定された列の一意の組み合わせに基づいて出力をサブディレクトリに整理します。
例えば
/path/to/output/color=red/fruit=apple/part-0000.parquet
/path/to/output/color=green/fruit=banana/part-0001.parquet
これにより、パーティション プルーニングによってクエリ パフォーマンスが向上します。
1 つのファイルに統合されません。
パーティションでは NULL 値が許可されます。
.mode("append") が使用されない限り、「追加」は行われません。
参考:パーティショニングを使用した Spark Write

正確な抜粋からの包括的かつ詳細な説明：
ディスクに書き出されるデータがソートされていることを確認するには、SparkがParquetテーブルに保存する際にデータを書き込む方法を考慮することが重要です。.sort()または.orderBy()メソッドはグローバルソートを適用しますが、特定の条件（例えば、スケーラブルではない.coalesce(1)による単一パーティションなど）が満たされない限り、最終出力ファイルでソートが維持されることは保証されません。
代わりに、分散 Spark 処理で行が書き出されるときにそれぞれのパーティション内でソートされるようにするための適切な方法は次のとおりです。
sortWithinPartitions("列名")
Apache Spark のドキュメントによると:
「sortWithinPartitions() は、各パーティションが指定された列でソートされていることを保証します。これは、ソートされたファイルを必要とする下流システムに役立ちます。」このメソッドは分散設定で効率的に動作し、グローバルソート（in.orderBy() や.sort() など）のパフォーマンスボトルネックを回避し、各出力パーティションにソートされたレコードが含まれることを保証します。これにより、一貫してソートされたデータという要件が満たされます。
したがって：
オプション A と B では、保存されたファイルの内容がソートされることは保証されません。
オプション C は、.coalesce(1) (単一パーティション) によってボトルネックが発生します。
オプション D はパーティション内でソートを正しく適用し、スケーラブルです。
参考: Databricks & Apache Spark 3.5 ドキュメント # DataFrame API # sortWithinPartitions()

DataFrameWriter を使用して既存のファイルを上書きする正しい方法は次のとおりです。
df.write.mode("上書き").json("ファイルへのパス")
オプション D も技術的には有効ですが、オプション A は最も簡潔で慣用的な PySpark 構文です。

正確な抜粋からの包括的かつ詳細な説明：
分散ワーカーからドライバーに情報 (最大値など) を集約する正しい方法は、reduce() やaggregate() などの RDD アクションを使用することです。
ドキュメントより:
「分散データに対してグローバル集計を実行するには、min/max/avg などの集計を収集するために、reduce() などのアクションが一般的に使用されます。」アキュムレータ (オプション B) は最大演算を直接サポートしておらず、このような分析には適していません。
ブロードキャスト (オプション C) は、ワーカーからデータを収集するのではなく、ワーカーに送信するために使用されます。
Spark UI (オプション D) は監視ツールであり、分析収集インターフェースではありません。
最終回答：A