PySparkとPandas データフレームの変換
Databricks で Apache Arrow を使用して、Apache Spark データフレーム をpandas データフレーム との間で変換する方法について説明します。
Apache Arrow と PyArrow
Apache Arrow は、JVMとPythonプロセス間でデータを効率的に転送するためにApache Sparkで使用されるインメモリ列指向データ形式です。 Pythonこれは、Pandas NumPyデータや データを扱う 開発者にとって有益です。ただし、その使用には、互換性を確保し、最大限のメリットを得るために、いくつかの小さな構成またはコード変更が必要です。
PyArrow は Apache Arrow の Python バインディングであり、Databricks Runtime にインストールされます。 各 Databricks Runtime バージョンで利用可能なPyArrowのバージョンに関する情報については、 Databricks Runtime リリースノートのバージョンと互換性を参照してください。
サポートされている SQL 型
すべての Spark SQL データ型は、TimestampTypeの ArrayType を除き、矢印ベースの変換でサポートされています。ネストされたStructTypeのMapTypeとArrayTypeは、PyArrow 2.0.0 以降を使用している場合にのみサポートされます。StructType は pandas.Seriesではなく pandas.DataFrame として表されます。
PySpark データフレームとPandas データフレームの変換
Arrow は、PySpark データフレーム を toPandas() でpandas データフレーム に変換するとき、および createDataFrame(pandas_df)を使用してpandasのデータフレームから PySpark データフレーム を作成するときに最適化として使用できます。
これらのメソッドで Arrow を使用するには、Spark 設定spark.sql.execution.arrow.pyspark.enabledを trueに設定します。この構成は、High Concurrencyクラスターと、 Unity Catalog 有効になっているワークスペースのユーザー分離クラスターを除き、デフォルトによって有効になります。
さらに、 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() は データフレーム 内のすべてのレコードをドライバー プログラムに収集するため、データの小さなサブセットに対して実行する必要があります。
さらに、すべての Spark データ型がサポートされているわけではなく、列の型がサポートされていない場合はエラーが発生する可能性があります。 createDataFrame()中にエラーが発生した場合、Spark は Arrow なしで データフレーム を作成します。