{"id":2359,"date":"2020-04-06T03:48:50","date_gmt":"2020-04-06T03:48:50","guid":{"rendered":"http:\/\/kusuaks7\/?p=1964"},"modified":"2023-12-20T14:28:59","modified_gmt":"2023-12-20T14:28:59","slug":"practical-spark-tips-for-data-scientists","status":"publish","type":"post","link":"https:\/\/www.experfy.com\/blog\/bigdata-cloud\/practical-spark-tips-for-data-scientists\/","title":{"rendered":"Practical Spark Tips for Data Scientists"},"content":{"rendered":"\t\t
\n\t\t\t\t\t\t
\n\t\t\t\t\t\t
\n\t\t\t\t\t
\n\t\t\t
\n\t\t\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tI know \u2014 Spark is sometimes frustrating to work with.<\/em><\/strong>\n\nAlthough sometimes we can manage our big data using tools like\u00a0Rapids<\/a>\u00a0or\u00a0Parallelization<\/a>,<\/em><\/strong>\u00a0there is no way around using Spark if you are working with Terabytes of data.\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tWhy? Because Spark gives memory errors a lot of times, and it is only when you genuinely work on big datasets with spark, would you be able to truly work with Spark.\n\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tThis post is going to be about \u2014 \u201cPractical Spark and memory management tips for Data Scientists.\u201d<\/strong>\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t

1. Map Side Joins<\/h2><\/h2>\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\"\"\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\t

alt=”Joining Dataframes” \/><\/p>

The syntax of joins in Spark is pretty similar to pandas:<\/p>

df3 = df1.join(df2, df1.column == df2.column,how='left'\n<\/code><\/pre>\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tBut I faced a problem. The df1 had around 1Billion rows while df2 had around 100 Rows. When I tried the above join, it didn\u2019t work and failed with memory exhausted errors after running for 20 minutes.\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tI was writing this code on a pretty big cluster with more than 400 executors with each executor having more than 4GB RAM. I was stumped as I tried to repartition my data frames using multiple schemes, but nothing seemed to work.\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tSo what should I do? Is Spark not able to work with a mere billion rows? Not Really. I just needed to use Map-side joins or broadcasting in Spark terminology.\n
from pyspark.sql.functions import broadcast\ndf3 = df1.join(broadcast(df2), df1.column == df2.column,how='left')\n<\/code><\/pre>\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tUsing the simple broadcasting code above, I was able to send the smaller df2 to all the nodes, and this didn\u2019t take a lot of time or memory. What happens in the backend is that a copy of df2 is sent to all the partitions and each partition uses that copy to do the join. That means that there is no data movement when it comes to df1, which is a lot bigger than df2.\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t
\n
2. Spark Cluster Configurations<\/h2><\/h2>\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\"\"\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\t
Set the Parallelism and worker nodes based on your task size<\/em><\/p>\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tWhat also made my life difficult while I was starting work with Spark was the way the Spark cluster needs to be configured. Your spark cluster might need a lot of custom configuration ad tuning based on the job you want to run.\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\t\nSome of the most important configurations and options are as follows:\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t
a. spark.sql.shuffle.partitions and spark.default.parallelism:<\/h3><\/h3>\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tspark.sql.shuffle.partitions configures the number of partitions to use when shuffling data for joins or aggregations. The spark.default.parallelism is the default number of partitions in RDDs returned by transformations like join, reduceByKey, and parallelize when not set by the user. The default value for these is 200.\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\t\nIn simple words, these set the degree of parallelism you want to have in your cluster.<\/em><\/strong>\n\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tIf you don\u2019t have a lot of data, the value of 200 is fine, but if you have huge data, you might want to increase these numbers. It also depends on the number of executors you have. My cluster was pretty big with 400 executors, so I kept this at 1200. A rule of thumb is to keep it as a multiple of the number of executors so that each executor ends up with multiple jobs.\n
sqlContext.setConf( \"spark.sql.shuffle.partitions\", 800)\nsqlContext.setConf( \"spark.default.parallelism\", 800)\n<\/code><\/pre>\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t
b. spark.sql.parquet.binaryAsString<\/h3><\/h3>\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tI was working with .parquet files in Spark, and most of my data columns were strings. But somehow whenever I loaded the data in Spark, the string columns got converted into binary format on which I was not able to use any string manipulation functions. The way I solved this was by using:\n
sqlContext.setConf(\"spark.sql.parquet.binaryAsString\",\"true\")\n<\/code><\/pre>\nThe above configuration converts the binary format to string while loading parquet files. Now it is a default configuration I set whenever I work with Spark.\t\t\t\t\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t
c. Yarn Configurations:<\/h3><\/h3>\t\t\t\t<\/div>\n\t\t\t\t<\/div>\n\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\tThere are other configurations that you might need to tune that define your cluster. But these need to be set up when the cluster is starting and are not as dynamic as the above ones. The few I want to put down here are for managing memory spills on the executor nodes. Sometimes the executor core gets a lot of work.\n