You probably understand just how painful data copying between DataFrames and DataFrames from the JVM can be when working with PySpark and other libraries such as DataFrames from Apache Pandas. This kind of data copying often takes place when doing data analysis with Apache Pandas prior to processing with Apache Spark or when visualizing data from Apache Spark in Apache DataFrames. Data copying was common and very costly because of row-by-row data communication between Apache Spark and Apache Python applications because of their JVM nature. Apache Arrow was developed to mitigate this challenge and optimize data communication. Apache Arrow is an open, cross-language, in-memory data format with analytics in mind. Unlike storing data in a form of Python objects or Java objects, Arrow data is organized in a common columnar memory format that can be shared directly across platforms. The format is native to Spark, Pandas, NumPy, and a whole range of analytics libraries such that data between them can be shared without any copying. The most valuable feature in Arrow is the columnar memory model it offers. The data can be processed in columns and not in rows. This feature makes it most useful in analytical tasks because it allows operations on the entire columns. Zero-copy data sharing can also be done in Arrow. What it refers to is the capability to share data between Python and the JVM without having to create objects in Python for all the data and then reassembling it in the Java world. Arrow can be easily integrated across all programming languages because it offers support in C/C++/Java/Python and R. That helps in mixed language data platforms such as Spark.  In the python world, Arrow is accessible through Apache PyArrow. PyArrow is a python extension package that provides a pythonic interface for Arrow data structures, essentially being the conduit between DataFrames in pandas and DataFrames in Spark. When PyArrow is enabled, DataFrames in pandas are automatically converted into Arrow Tables that Spark can efficiently handle in the JVM world, skipping the usual slow data path that involves python object serialization. Legacy inefficient  way: How PyArrow Internally Operates spark.conf.set("spark.sql.execution.arrow.enabled", "true") spark.conf.set("spark.sql.execution.arrow.enabled", "true") Alternatively, you can enable it globally by adding the following entry to your spark-defaults.conf file.     spark.sql.execution.arrow.enabled=true spark.sql.execution.arrow.enabled=true In PySpark, using Apache Arrow requires manual enablement. Apache Arrow Execution mode is disabled by default to work along with existing systems in an ideal way. Apache Arrow Execution mode can be enabled using Spark Configuration. After this feature is enabled, Spark will automatically try to use Arrow in any conversions from Pandas to Spark data frames that it can handle.  To appreciate how it affects overall performance, note that a simple experiment where a Pandas DataFrame is transformed into a Spark DataFrame can be considered. For the initial case, the Arrow library is not functioning. When this happens, Spark promotes each Pandas row into a Python object, sends it in serialized form to the JVM, where it gets reconstructed into Spark's internal representation. import numpy as np
import pandas as pd
import time
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("PyArrow_Test").getOrCreate()

pdf = pd.DataFrame(np.random.rand(100000, 3))

start_time = time.time()
df1 = spark.createDataFrame(pdf)
end_time = time.time()

print("Time taken without Apache Arrow:", end_time - start_time) import numpy as np
import pandas as pd
import time
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("PyArrow_Test").getOrCreate()

pdf = pd.DataFrame(np.random.rand(100000, 3))

start_time = time.time()
df1 = spark.createDataFrame(pdf)
end_time = time.time()

print("Time taken without Apache Arrow:", end_time - start_time) In the later case, Arrow is activated. Instead of serializing row-wise, Spark converts the Pandas DataFrame into a columnar Arrow Table, which is transferred in compact binary format to the JVM. spark.conf.set("spark.sql.execution.arrow.enabled", "true")

start_time = time.time()
df2 = spark.createDataFrame(pdf)
end_time = time.time()

print("Time taken with Apache Arrow:", end_time - start_time) spark.conf.set("spark.sql.execution.arrow.enabled", "true")

start_time = time.time()
df2 = spark.createDataFrame(pdf)
end_time = time.time()

print("Time taken with Apache Arrow:", end_time - start_time) Under normal circumstances, of course, the contrast is marked. A conversion that might take several seconds to perform without Arrow can be accomplished in a few tens of milliseconds with Arrow optimization switched on. This advantage grows with larger datasets. Internally, with Arrow enabled, the data conversion flow looks quite different. The Pandas DataFrame is mapped first to an Arrow Table from PyArrow, where each column of data is represented as a chunk of memory. The result remains an Arrow Table and gets transferred directly to the JVM, where Spark further translates it into an internal columnar format. There isn't an object created in Python in this scenario either. Without the use of Arrow, the process incurs significantly high costs. Each Pandas dataframe row is converted into Python objects, which are sent across the Python JVM, deserialized into Java objects, and then scanned again for the Spark schema inference. The creation and reclamation of objects are the root causes of the poor performance of the process. The advantage of Apache Arrow goes beyond performance speed. Memory management efficiency improves because Apache Arrow prevents creating millions of Python or Java objects when performing memory transfers. Processing speed also increases because Apache Arrow technologies use vectorized processing, which enhances the processing speed of the CPU. The data transfer process also becomes easier and easier to predict because of Apache Arrow. Nevertheless, Arrow is far from a magic solution. Not all types of data are handled there, and for very complicated nested data formats and custom objects, it's possible to resort to a non-Arrow solution. At the same time, arrow-based processing may put additional memory pressure due to buffer materialization of bulky columnar objects. Consequently, it is important that arrow-enabled processing be tested with realistic volumes of data. In today’s PySpark applications, it’s high time to treat Apache Arrow as an optimization that must be made automatically whenever there’s an interaction between Pandas and Spark. The use of Apache Arrow will get rid of one of the most common bottlenecks in Spark applications developed using Python. Advanced Optimizations and Current Best Practices In recent versions of Spark, Arrow is also utilized internally for Pandas UDFs. Pandas UDFs work with Arrow Batches as opposed to rows, which allows vectorized execution that can be several orders of magnitude faster than the standard Python UDFs. Here, it can be noted that Arrow is a critical component for scalable feature engineering and transformation in a Python-based system.  Another consideration is batch size tuning. The Arrow is batch-aware, with the number of batches governed by spark.sql.execution.arrow.maxRecordsPerBatch. Having a larger number of records per batch will enhance throughput but will result in higher memory usage. Small batches are recommended when memory is constrained. for example : spark.sql.execution.arrow.maxRecordsPerBatch spark.sql.execution.arrow.maxRecordsPerBatch=10000 spark.sql.execution.arrow.maxRecordsPerBatch=10000 it is also crucial to take cognizance of fallback mechanisms. Spark will automatically fall back on the non-Arrow solution when there are data types that are not supported like nested maps. It is essential to take cognizance of Spark logs to ensure that Arrow is being utilized. Benefits of Using Apache Arrow Apache Arrow provides performance gains with scaled-down overheads of serialization and facilitating vectorized computations. Apache Arrow consumes minimal memory by eliminating the formation of millions of temporary objects in both Java and Python. Apache Arrow provides interoperability and promotes a common memory model between analytical systems.  Under the new Spark architecture, the uses of Arrow are not limited to the conversion from Pandas conversion as an optimization; rather, it serves as the building block for enabling scalable Python execution, optimized UDFs, and the future vision of the Spark client-server architecture. One cannot emphasize the importance of considering Arrow as a first-class dependency in every full-scale application of the PySpark library interfaced with the Pandas library. Real world use cases : Financial Services Apache Arrow is used at large banks to accelerate risk calculations where Pandas-based feature engineering feeds Spark models that handle billions of trades with strict latency budgets. Fintech & Payments: Payment platforms depend on Arrow-enabled Pandas UDFs to compute real-time fraud scores, enabling Python models to run at Spark scale-without serialization bottlenecks. E Commerce and Retail: The retail analytics groups employ Arrow to transfer customer clickstream data from Pandas to Spark for fast experimentation before applying recommendations to the entire catalog. Healthcare and Life Sciences : The genomics pipelines make use of the Arrow library for efficiently transferring large matrix-like data from Python-based statistical analysis to Sparks-driven large-scale data processing. Telecommunications : Telecommunication companies employ Arrow to analyze their network telemetry data, translating Anomaly Detection in Pandas results into Spark jobs processing petabytes of signal data. Telecommunications companies employ Arrow in analyzing their network telemetry data, translating Anomaly Detection Ad Tech & Marketing Industry: The marketing platforms use Arrow to vectorize attribution models. Using this process, Python underlying feature transformation abilities execute quickly for huge impression datasets. Machine Learning Platforms & Online Tools Feature stores integrate Arrow into the data exchange functionality for handling data between offline Spark training pipelines and online inference processes involving Python.  Log Analytics & Observability Engineers employ Arrow to swiftly log enrichment, using Python parsers and Spark aggregation jobs that do not entail high levels of serialization costs. Data Warehousing & BI : In most The Cloud warehouses make the result sets of Arrow based queries available to Python clients, so it becomes easier for Spark and Pandas users to retrieve large analytical datasets at a low data transfer cost. Scientific Computing and Research : In today's Research institutions employ Arrow for the integration of the numpy, pandas, and Spark libraries and perform scalable volumetric computations in shared in-memory models. Why Apache Arrow Wins in Production : In each of these scenarios, the same pattern holds. Arrow eliminates serialization from the critical path, reduces memory amplification, and allows for vectorization. This directly correlates to improvement in the aspects of latency, throughput, and predictability of resource usage.  More importantly, Arrow empowers data engineering teams to express high-quality data pipelines using Python while making system efficiency and developer productivity hand-in-hand goals, which has contributed significantly to the prominence of the technology named Arrow in the current state of data platforms.

Apache Arrow: Optimizing PySpark Applications

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

Mastering IoT Data Surges with Spark Streaming Backpressure

The Noonification: How Often Do NFTs Pass The Howey Test? (1/13/2023)

Darwin's Hybrid Intelligence to Align AI & Human Goals for Startups & VCs

The Noonification: White Man (11/26/2022)

The Noonification: The Metaverse is a Sh*tshow (11/2/2022)

100 Days of AI Day 1: From Newsletter to Podcast, Leveraging AI for Audio Transformation

Mastering IoT Data Surges with Spark Streaming Backpressure

The Noonification: How Often Do NFTs Pass The Howey Test? (1/13/2023)

Darwin's Hybrid Intelligence to Align AI & Human Goals for Startups & VCs

The Noonification: White Man (11/26/2022)

The Noonification: The Metaverse is a Sh*tshow (11/2/2022)

100 Days of AI Day 1: From Newsletter to Podcast, Leveraging AI for Audio Transformation

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps