PYSPARK FHASH CODE
We can run the following code to use a custom paritioner: So we can only use this function with RDD class.Īs partitionBy function requires data to be in key/value format, we need to also transform our data. To address the above issue, we can create a customised partitioning function.Īt the moment in PySpark (my Spark version is 2.3.3), we cannot specify partition function in repartition function. However, what if the hashing algorithm generates the same hash code/number? Use partitionBy function The output shows that each country’s data is now located in the same partition: So if we increate the partition number to 5. In this way, the chance for allocating each different value to different partition is higher. Well, the first thing we can try is to increase the partition number. Allocate one partition for each key valueįor the above example, if we want to allocate one partition for each Country (CN, US, AU), what should we do? There is also one partition with empty content as no records are allocated to that partition. This output is consistent with the previous one as record ID 1,4,7,10 are allocated to one partition while the others are allocated to another question. Udf_portable_hash = udf(lambda str: portable_hash(str))ĭf = df.withColumn("Hash#", udf_portable_hash(df.Country))ĭf = df.withColumn("Partition#", df % numPartitions) We can verify this by using the following code to calculate the hash For different country code, it may be allocated into the same partition number. Why? Because repartition function by default uses hash partitioning. You may expect that each partition includes data for each Country but that is not the case. Let’s repartition the data to three partitions only by Country column.ĭf = df.repartition(numPartitions, "Country") Records are divided into 8 partitions as 8 worker threads were configured. The output from print_partitions function is shown below: In the above code, print_partitions function will print out all the details about the RDD partitions including the rows in each partition. # Create Spark session with Hive supported.ĭata.append(".format(df.rdd.partitioner)) Let’s first create a data frame using the following code: Walkthrough with data Create a sample data frame The partitionBy function is defined as the following:ĭef partitionBy(self, numPartitions, partitionFunc=portable_hash)īy default, the partition function is portable_hash. Partitioner then uses this partition function to generate the partition number for each keys.
0 kommentar(er)
