0

我正在尝试使用 PySpark 从 RDS MySQL 实例中读取表。这是一个巨大的表，因此我想通过使用分区概念来并行化读取操作。该表没有用于查找分区数的数字列。相反，它有一个时间戳列（即日期时间类型）。

我通过检索时间戳列的最小值和最大值找到了下限和上限。但是，我不确定是否有标准公式可以动态找出分区数。这是我目前正在做的事情（硬编码 numPartititons 参数的值）：

select_sql = "SELECT {} FROM {}".format(columns, table)
partition_info = {'partition_column': 'col1', 
                  'lower_bound': '<result of min(col1)>', 
                  'upper_bound': '<result of max(col1)>', 
                  'num_partitions': '10'}
read_df = spark.read.format("jdbc") \
        .option("driver", driver) \
        .option("url", url) \
        .option("dbtable", select_sql) \
        .option("user", user) \
        .option("password", password) \
        .option("useSSL", False) \
        .option("partitionColumn", partition_info['partition_column']) \
        .option("lowerBound", partition_info['lower_bound'])) \
        .option("upperBound", partition_info['upper_bound'])) \
        .option("numPartitions", partition_info['num_partitions']) \
        .load()

请向我建议一个可行的解决方案/您的方法。谢谢

4

1 回答 1

0

如何设置numPartitions取决于您的集群的定义。这里没有正确或错误或自动设置。只要您了解 、 、 、 以及可能大量基准测试背后的逻辑partitionColumn，lowerBound您upperBound就numPartitions可以决定什么是正确的数字。

Pyspark - df.cache().count() 永远运行

partitionColumn、lowerBound、upperBound、numPartitions参数是什么意思？

于 2021-05-19T22:14:49.557 回答