使用自定义分割器在Pyspark中分割数据帧

Question

查找有关在 Pyspark 中使用自定义分区程序的一些信息。我有一个数据帧，其中包含各个国家/地区的国家/地区数据。因此，如果我在国家/地区列上重新分区，它会将我的数据分发到 n 个分区中，并将类似的国家/地区数据保留到特定分区。当我看到使用 glom（） 方法时，这是创建一个倾斜分区数据。

一些国家，如美国和中国，在特定的数据帧中拥有大量数据。我想重新分区我的数据帧，这样如果国家是美国和中国，那么它将进一步拆分为大约 10 个分区，否则对其他国家（如印度、泰国、澳大利亚等）保持分区相同。我们可以在 Pyspark 代码中扩展分区程序类吗？

我在下面的链接中读到了这一点，我们可以在scalaSpark应用程序中扩展scalapartitioner类，并可以修改partitionerclass以使用自定义逻辑根据需求重新划分数据。就像我所做的一样，请帮助在Pyspark中实现这个解决方案。请参阅下面的链接。按列分区但保持固定分区计数的有效方法是什么？

我使用的是Spark版本2.3.0.2下面是我的Dataframe结构：

datadf= spark.sql("""
    SELECT    
        ID_NUMBER ,SENDER_NAME ,SENDER_ADDRESS ,REGION_CODE ,COUNTRY_CODE
    from udb.sometable
""");

输入的数据包含六个国家的数据，如AUS、IND、THARUS、CHNUSACHN和USA具有歪斜数据。

因此，如果我对COUNTRY_code执行重新分区，两个分区包含大量数据，而其他分区则很好。我使用glom（）方法检查了这一点。

newdf = datadf.repartition("COUNTRY_CODE")

from pyspark.sql import SparkSession
from pyspark.sql import  HiveContext, DataFrameWriter, DataFrame

newDF = datadf.repartitionByRange(3,"COUNTRY_CODE","USA")

我正在尝试将数据重新分区为仅针对美国和中国国家/地区的另外 3 个分区，并希望将其他国家/地区的数据保留在单个分区中。

This is what I am expecting 
AUS- one partition
IND- one partition
THA- one partition
RUS- one partition
CHN- three partition
USA- three partition

回溯（最近一次调用）：文件“”，第1行，在文件“/usr/hdp/current/spark2 client/python/pysark/sql/dataframe.py”，第1182行，在getattr“%s”对象中没有属性“%s”%（self.class.name，name））AttributeError:“dataframe”对象没有属性“repartitionByRange”

Answer 1

没有在PySpark上应用用户定义的分区器的直接方法，捷径是创建一个带有UDF的新列，根据业务逻辑为每个记录分配一个分区ID。并使用新列进行分区，这样数据可以均匀分布。

numPartitions= 3
df = df.withColumn("Hash#", udf_country_hash(df['Country']))
df = df.withColumn("Partition#", df["Hash#"] % numPartitions)
df.repartition(numPartitions, "Partition#")

请查看在线版本的代码@https://databricks-prod-cloudfront.cloud.databricks.com/public/4027ec902e239c93eaaa8714f173bcfc/8963851468310921/2231943684776180/5846184720595634/latest.html

根据我的经验，将DataFrame转换为RDD，然后再转换回DataFrame是一项开销很大的操作，最好避免。

Answer 2

结构化API中没有自定义分区程序，因此为了使用自定义分区程序，您需要下拉到RDD API。简单的3个步骤如下：

1. 将结构化 API 转换为 RDD API

dataRDD = dataDF.rdd

import random

# Extract key from Row object
dataRDD = dataRDD.map(lambda r: (r[0], r))

def partitioner(key):
    if key == "CHN":
        return random.randint(1, 10)
    elif key == "USA":
        return random.randint(11, 20)
    else:
        # distinctCountryDict is a dict mapping distinct countries to distinct integers
        # these distinct integers should not overlap with range(1, 20)
        return distinctCountryDict[key]

numPartitions = 100
dataRDD = dataRDD.partitionBy(numPartitions, partitioner)

# Remove key extracted previously
dataRDD = dataRDD.map(lambda r: r[1])

dataDF = dataRDD.toDF()

这样，您可以在结构化API中获得两个世界的最佳状态，Spark类型和优化的物理计划，以及在低级RDD API中的自定义分区器。只有在绝对必要时，我们才使用低级API。

Answer 3

尝试这样的事情与哈希：

newDf = oldDf.repartition(N, $"col1", $"coln")

或者对于测距方法:

newDF = oldDF.repartitionByRange(N, $"col1", $"coln")

DF还没有自定义分区。

在你的情况下，我会去散列，但没有保证。

但是，如果您的数据有偏差，您可能需要一些额外的工作，例如用于分区的 2 列是最简单的方法。

例如，现有或新列 - 在这种情况下，对给定国家/地区应用分组的列，例如 1 ..N 和两个列上的分区。

对于有许多分组的国家，您可以得到N个合成子分区；对于具有低基数的其他，仅具有1个这样的组号。不太难。两个分区都可以使用多于1个列。

在我看来，分区的统一数字填充需要付出很多努力，而且无法真正实现，但是像这里这样的下一个最佳方法就足够了。相当于一定程度的自定义分区。

否则，在 DF 上使用 .withColumn，您可以使用这些规则模拟自定义分区并填充新的 DF 列，然后应用 repartitionByRange。也不难。