PySpark と pandas DataFrame 間で変換する
Azure Databricks で Apache Arrow を使用して、Apache Spark DataFrame と pandas DataFrame の間で相互に変換する方法について説明します。
Apache Arrow と PyArrow
Apache Arrow は、JVM と Python のプロセス間で効率的にデータを転送するために Apache Spark で使用されるインメモリの列指向データ形式です。 これは、pandas と NumPy データを使用する Python 開発者に役立ちます。 ただし、その使用にあたっては、互換性を確保して最大限のメリットを得るために、構成またはコードにいくつかの軽微な変更が必要になります。
PyArrow は Apache Arrow の Python バインドであり、Databricks Runtime にインストールされます。 各 Databricks Runtime バージョンで使用できる PyArrow のバージョンについては、Databricks Runtime リリース ノートのバージョンと互換性 を参照してください。
サポートされている SQL の型
ArrayType の TimestampType を除くすべての Spark SQL データ型が、Arrow ベースの変換でサポートされています。 入れ子になった MapType の ArrayType と StructType は、PyArrow 2.0.0 以降を使用する場合にのみサポートされます。 StructType は、pandas.DataFrame ではなく pandas.Series として表されます。
PySpark DataFrame と pandas DataFrame の間の変換
toPandas() を使用して PySpark DataFrame を pandas DataFrame に変換する場合と、createDataFrame(pandas_df) を使用して pandas DataFrame から PySpark DataFrame を作成する場合の最適化として Arrow を使用できます。
これらのメソッドに Arrow を使用するには、Spark 構成spark.sql.execution.arrow.pyspark.enabled を trueに設定します。 この構成は、高コンカレンシー クラスターと、Unity カタログが有効になっているワークスペース内のユーザー分離クラスターを除き、既定で有効になります。
また、spark.sql.execution.arrow.pyspark.enabled で有効化された最適化は、Spark での計算の前にエラーが発生した場合、Arrow 以外の実装にフォールバックすることがあります。 この動作は、Spark 構成 spark.sql.execution.arrow.pyspark.fallback.enabled を使用して制御できます。
例
import numpy as np
import pandas as pd
# Enable Arrow-based columnar data transfers
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
# Generate a pandas DataFrame
pdf = pd.DataFrame(np.random.rand(100, 3))
# Create a Spark DataFrame from a pandas DataFrame using Arrow
df = spark.createDataFrame(pdf)
# Convert the Spark DataFrame back to a pandas DataFrame using Arrow
result_pdf = df.select("*").toPandas()
Arrow の最適化を使用して、Arrow が有効になっていない場合と同じ結果を生成します。 Arrow を使用した場合でも、toPandas() は、結果として DataFrame 内のすべてのレコードをドライバー プログラムに収集し、データの小規模なサブセットに対して実行されます。
さらに、すべての Spark データ型がサポートされているわけではありません。列にサポートされていない型がある場合はエラーが発生する可能性があります。 createDataFrame() でエラーが発生した場合、Spark によって Arrow なしに DataFrame が作成されます。