Following up to my Scaling Python for Data Science using Spark<\/a> post where I mentioned Spark 2.3 introducing Vectorized UDFs, I’m using the same Data (from NYC yellow cabs) with this code:<\/p>\n Pandas Scalar (the default; as opposed to grouped map) UDFs operate on pandas.Series<\/a> objects for both input and output, hence the .dt<\/span> call chain as opposed to directly calling strftime on a python datetime object. The entire functionality is dependent on using PyArrow (>= 0.8.0).<\/p>\n Expect errors to crop up as this functionality is new. I have seen a fair share of memory leaks and casting errors causing my jobs to fail during testing.<\/p>\n Running the job above shows some new items in the Spark UI (DAG) and explain plan:<\/p>\n Note the addition of ArrowEvalPython<\/p>\n To jog your memory, PySpark SQL took 17 seconds to count the distinct epoch timestamps, and regular Python UDFs took over 10 minutes (610 seconds).<\/p>\n Much to my dismay, the performance of my contrived test was in line with Python UDFs, not Spark SQL with a runtime of 9-10 minutes.<\/p>\n I’ll update this post [hopefully] as I get more information.<\/p>\n","protected":false},"excerpt":{"rendered":" Since Spark 2.3 was officially released 2\/28\/18, I wanted to check the performance of the new Vectorized Pandas UDFs using Apache Arrow. Following up to my Scaling Python for Data Science using Spark post where I mentioned Spark 2.3 introducing Vectorized UDFs, I’m using the same Data (from NYC yellow cabs) with this code: from… Continue readingfrom pyspark.sql import functions as F\r\nfrom pyspark.sql.functions import pandas_udf, PandasUDFType\r\nfrom pyspark.sql.types import *\r\nimport pandas as pd\r\n\r\ndf = spark.read\\\r\n .option(\"header\", \"true\")\\\r\n .option(\"inferSchema\", \"true\")\\\r\n .csv(\"yellow_tripdata_2017-06.csv\")\r\n\r\ndef timestamp_to_epoch(t):\r\n return t.dt.strftime(\"%s\").apply(str) # <-- pandas.Series calls\r\n\r\nf_timestamp_copy = pandas_udf(timestamp_to_epoch, returnType=StringType())\r\ndf = df.withColumn(\"timestamp_copy\", f_timestamp_copy(F.col(\"tpep_pickup_datetime\")))\r\ndf.select('timestamp_copy').distinct().count()\r\n\r\n# s = pd.Series({'ds': pd.Timestamp('2018-03-03 04:31:19')})\r\n# timestamp_to_epoch(s)\r\n## ds 1520080279\r\n## dtype: object<\/pre>\n![\"https:\/\/www.bandegraphix.com\/DecalImages\/531-caution-m.png\"](\"https:\/\/www.bandegraphix.com\/DecalImages\/531-caution-m.png\")
<\/p>\n![\"\"](\"http:\/\/garrens.com\/blog\/wp-content\/uploads\/2018\/03\/Screen-Shot-2018-03-03-at-8.20.20-PM.png\")
<\/a> ![\"\"](\"http:\/\/garrens.com\/blog\/wp-content\/uploads\/2018\/03\/Screen-Shot-2018-03-03-at-8.20.02-PM.png\")
<\/a><\/p>\nWhat’s the performance like?!<\/strong><\/h3>\n