![\"\"](\"http:\/\/garrens.com\/blog\/wp-content\/uploads\/2018\/01\/apache-spark.jpg\")
<\/a><\/p>\nPython is the de facto language of Data Science & Engineering. (IMHO R is grand for statisticians, but Python is for the rest of us.)<\/p>\n
As a prominent language in the field, it only makes sense that Apache Spark supports it with Python specific APIs. Spark makes it so easy to use Python<\/strong>\u00a0 that it can actually run slow (up to 30-40x slower<\/span> than it should) due to naive missteps<\/strong>.<\/p>\n Remember: Spark is written in Scala primarily to leverage JVM performance characteristics, but also has APIs in Python, Java, and R<\/p><\/blockquote>\n Even with the great (and rapidly growing!) support for Python on Spark (henceforth PySpark), there are some critical topics that need to be addressed. Do any of these packages look familiar, Python devs?<\/p>\n Thought so. \ud83d\ude42<\/p>\n Python is fantastic for Initial and Exploratory Data Analysis (IDA\/EDA) plus training models and all sorts of other awesome stuff. However, it can’t scale to larger data sizes\/parameters due to 2 primary issues:<\/p>\n 1) it’s natively single-threaded* due to global interpreter lock (GIL) – out of scope for this post<\/p>\n 2) It is not distributed** to multiple machines, so even if it were multi-threaded, it is still bound by one machine’s resources (CPU cores + memory + I\/O + network)<\/p>\n * yes, there are different implementations of Python (CPython, Jython, Pypy, etc) that may or may not have mult-threaded\/concurrency supported. We’re focusing on the default and predominantly used CPython implementation<\/p>\n ** It can<\/em> be distributed with a hot [relatively] new package “dask<\/a>” that came out in 2014. We’re sticking with PySpark for this post, but the more you know… AWS does offer a single node (x1e.32xlarge) with 128 vCPUs and ~3.9TB of RAM. If you have to ask how much it costs, you can’t afford it. \ud83d\ude09<\/p>\n When people think of Spark, one of the first things that comes to mind is the age-old RDD. It is after all the original data structure API that brought Spark to prominence. RDDs still underpin Spark, however since version 2.0<\/em>, the preferred data structure API for Python has become DataFrames<\/a><\/strong> (corollary in Scala\/Java is the Datasets<\/a> API)<\/p>\n DataFrames<\/em> are under the [massive] umbrella that is “Spark SQL.” DataFrames are preferred to RDDs for two optimization-related reasons; the Catalyst Optimizer <\/strong><\/a>and Tungsten<\/a>. <\/strong>The short explanation of what Catalyst and Tungsten do for DataFrames is automatically handling optimizations that were too low-level for RDDs. The higher level Spark SQL abstractions afforded these APIs with more introspection and greater execution flexibility, thus increasing performance for DataFrames substantially over RDDs<\/strong>. For PySpark users, the performance difference is even greater<\/em> than for Scala\/Java Spark users, because actions done using PySpark SQL APIs directly (e.g. trimming a string, summing an aggregate, regex, etc) are actually executed in the JVM directly rather than in a Python subprocess like RDDs require.<\/p>\n When you’re using DataFrames, be sure to use the baked-in Spark SQL functions (also compatible with hive functions), because they all run optimized in the JVM: https:\/\/spark.apache.org\/docs\/latest\/api\/python\/pyspark.sql.html#module-pyspark.sql.functions<\/a><\/p>\n Moral of the story: If you’re using RDDs (especially with PySpark!), evaluate and try to transition to DataFrames. If you’re not using RDDs yet, try DataFrames first.<\/span><\/p>\n This section wouldn’t be complete without a silly [but truthful] benchmark:<\/p>\n You probably can,<\/em> but just because you can doesn’t mean you should.<\/span><\/em><\/p>\n What I mean by that is when you start mixing Python native code (such as Pandas) with Spark, two bad things happen [usually]. Firstly, your performance nose dives, because as I alluded to briefly above,\u00a0 any code that runs natively in Python (as opposed to using the Spark SQL APIs) requires [de]serialization to\/from Python\/JVM and execution in Python subprocesses. More info here<\/a>, which is a bit old by Spark standards, but any Spark-related wisdom by Holden Karau is great.<\/p>\n The Spark SQL DataFrame API only goes so far (it goes very far FWIW). So maybe you’re tempted to write a UDF (User Defined Function) to extend Spark’s functionality for your use case.<\/p>\n Before you write a UDF that uses Python-specific APIs (not from PySpark), have a look at this simple example and its implications.<\/p>\n We want to convert timestamps in a column to seconds since epoch and count the distinct outputs (yes, this part is purely for benchmarking).<\/p>\n Yes, you read that right. PySpark SQL APIs are 30-40 TIMES<\/span> FASTER (!!!)<\/strong><\/p>\n UDF with Python-Specific APIs: 610 seconds Fortunately for you, Spark 2.3 solves the issue – partly – by introducing an impressive feature allowing vectorization of Python UDFS, which you can read about more here<\/a>. Vectorized UDFs for PySpark<\/strong> can again massively improve performance.<\/strong><\/p>\n I’m a sucker for contrived benchmarks that make my case so clearly. \ud83d\ude00<\/p>\n Here’s a na\u00efve way to build a Spark DataFrame using Pandas first. Note this not only takes over 4 minutes to run, it also uses ~3GB of memory (!!).<\/p>\n On the contrary, the same line count using the Spark SQL APIs directly (5 seconds, negligible memory impact):<\/p>\n If you’re going to use Python-native packages with Spark, be mindful of the [unintended] consequences, particularly as it relates to your jobs’ performance.<\/p>\n For minor version changes, such as 2.1 to 2.3, many APIs are backwards compatible. Major version changes, such as 1.6 to 2.0 should definitely anticipate backwards incompatible changes. Structured Streaming did go through some fairly significant changes from 2.0.0 to 2.2.0 when it transitioned from experimental to production-ready respectively.<\/p>\n Special Thanks<\/strong><\/p>\n Thank you Wes Hoffman<\/a>, Nikki Haas<\/a> and Eric Lambert<\/a> for your feedback and suggestions on relevant topics!<\/p>\n Further reading:<\/p>\n https:\/\/databricks.com\/blog\/2016\/07\/14\/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html<\/p>\n https:\/\/databricks.com\/blog\/2017\/10\/30\/introducing-vectorized-udfs-for-pyspark.html<\/p>\n","protected":false},"excerpt":{"rendered":" Python is the de facto language of Data Science & Engineering. (IMHO R is grand for statisticians, but Python is for the rest of us.) As a prominent language in the field, it only makes sense that Apache Spark supports it with Python specific APIs. Spark makes it so easy to use Python\u00a0 that it… Continue readingpandas (aka \"pd\")\r\nnumpy (aka \"np\")\r\nscipy\r\nscikit-learn\r\nkeras\r\ntensorflow\r\nsqlalchemy<\/pre>\n
No, don’t use RDDs; use [Spark] SQL!
\n<\/strong><\/h2>\nNote: Dataset is the Scala and Java API that is strongly typed.\r\n<\/em>DataFrame is literally an alias for Dataset[Row] in Scala\r\nDataFrame does not exist in Java (except as Dataset[Row])\r\n\r\n*DataFrame is the only Spark SQL Data Structure API for Python, \r\nbecause Python is dynamically typed.*<\/pre>\n
%%time # RDD\r\nlines = sc.textFile(\"yellow_tripdata_2017-06.csv\")\r\nlineLengths = lines.map(lambda s: len(s))\r\ntotalLength = lineLengths.reduce(lambda a, b: a + b)\r\nprint(totalLength) # => 832591506\r\n\r\nCPU times: user 9.66 ms, sys: 2.99 ms, total: 12.6 ms\r\nWall time: 7.7 s<\/strong>\r\n\r\n%%time # DataFrame\r\nlinesDF = spark.read.text(\"yellow_tripdata_2017-06.csv\")\r\nprint(\r\n linesDF.selectExpr(\r\n \"sum(length(value)) as slen\")\r\n .collect()[0]['slen']\r\n) # => 832591506\r\n\r\nCPU times: user 4.76 ms, sys: 1.74 ms, total: 6.5 ms\r\nWall time: 3.84 s<\/strong><\/pre>\n
All hope is not lost on “strong typing”-like features for PySpark<\/h3>\n
Python being dynamically typed is a double edged sword, because by being dynamically typed (therefore forced to only use DataFrames, not Datasets), PySpark DataFrames also natively support accessing variables\/columns by name directly. For example…<\/h3>\n
linesDF = spark.read\r\n .option(\"header\", \"true\")\r\n .option(\"inferSchema\",\"true\")\r\n .csv(\"yellow_tripdata_2017-06.csv\")\r\nlinesDF.agg(\r\n pyspark.sql.functions.sum(\r\n linesDF.passenger_count # <---- HERE\r\n )\r\n)\r\n.collect()[0]['sum(passenger_count)']<\/pre>\n
I want to use [insert awesome package here] with PySpark<\/h2>\n
Example<\/h4>\n
Using UDFs with Python-specific APIs<\/h5>\n
timestamp_to_epoch = F.udf(lambda t: int(t.strftime(\"%s\")))\r\n%%time\r\ndf.select(timestamp_to_epoch(df.tpep_pickup_datetime)).distinct().count()\r\nCPU times: user 44.8 ms, sys: 18.3 ms, total: 63.1 ms\r\nWall time: 10min 10s<\/strong><\/span>\r\nOut[46]: 2340959<\/pre>\nUsing PySpark SQL APIs<\/h5>\n
%%time\r\ndf.select(F.unix_timestamp(df.tpep_pickup_datetime)).distinct().count()\r\nCPU times: user 2.67 ms, sys: 1.15 ms, total: 3.82 ms\r\nWall time: 16.5 s<\/strong><\/span>\r\nOut[47]: 2340959<\/pre>\n
\n<\/strong>PySpark SQL API: 17 seconds<\/strong><\/p>\n%%time\r\npdf = pd.read_table(\"yellow_tripdata_2017-06.csv\")\r\nspark.createDataFrame(pdf).count()\r\nCPU times: user 4min 2s, sys: 3.33 s, total: 4min 6s\r\nWall time: 4min 10s<\/strong><\/pre>\n
%%time\r\nspark.read.csv(\"yellow_tripdata_2017-06.csv\").count()\r\nCPU times: user 1.63 ms, sys: 1.15 ms, total: 2.77 ms\r\nWall time: 4.57 s<\/strong><\/pre>\n
Special note about Backwards Compatibility<\/h6>\n