如何设计查找下限

Question

为了在spark中加载和划分传入的数据，我使用了以下语法。

val dataframe = spark.read.format("jdbc")
          .option("url", url)
          .option("driver", driver)
          .option("user", user)
          .option("password", password)
          .option("dbtable", query)
          .option("partitionColumn", partitionColumn)
          .option("lowerBound", lowerBound_value)
          .option("upperBound", upperBound_value)
          .option("numPartitions", numPartitions)
          .option("fetchsize", 15000)
          .load()

参数< code>partitionColumn 、< code>lowerBound 、< code>upperBound 、< code>numPartitions用于优化作业的性能。

我有一个包含 1000 条记录的表

当数据较少时，上述设计效果很好。但我有一个场景如下。

我有一个包含 2030 亿条记录的表，其中没有包含唯一/串行整数的整数列。然后有一个日期列，其数据分布在 5 年，即 2016-2021 年。为了更快地移动数据，我每次都会移动每年一个月的数据。这是我正在使用的查询：

val query = s"(select * from table where date_column >= '${YearMonth.of(year.toInt, month).atDay(1).toString} and date_time <= '${YearMonth.of(year.toInt, month).atEndOfMonth().toString} 23:59:59.999') as datadf"

因此，上面的查询变成：＜code＞select*from table where date_column

这是对我的循环的粗略描述：

(2016 to 2021) { year =>
   (1 to 12) { month =>
           val query = s"(select * from table where date_column >= '${YearMonth.of(year.toInt, month).atDay(1).toString} and date_time <= '${YearMonth.of(year.toInt, month).atEndOfMonth().toString} 23:59:59.999') as datadf"
           val dataframe = spark.read.format("jdbc")
              .option("url", url)
              .option("driver", driver)
              .option("user", user)
              .option("password", password)
              .option("dbtable", query)
              .option("partitionColumn", partitionColumn)
              .option("lowerBound", lowerBound_value)
              .option("upperBound", upperBound_value)
              .option("numPartitions", numPartitions)
              .option("fetchsize", 15000)
              .load()
   }
}

为了找出界限，我使用了以下月份和年份的相同过滤器：

val bounds = spark.read.format("jdbc")
          .option("url", url)
          .option("driver", driver)
          .option("user", user)
          .option("password", password)
          .option("dbtable", "(select min(partitionColumn) as mn, max(partitionColum) as from tablename where date_column >= '${YearMonth.of(year.toInt, month).atDay(1).toString} and date_time <= '${YearMonth.of(year.toInt, month).atEndOfMonth().toString} 23:59:59.999') as boundsDF")
          .load()

val lowerBound_value = bounds.select("mn").head.getInt(0)
val upperBound_value = bounds.select("mx").head.getInt(0)

问题在于查找过滤数据的下限和上限。由于数据量巨大，运行最少的查询

我尝试在那里给出随机值，但是在任务运行时观察到分区中的数据倾斜。

必须给出then partitionColumn的最小值和最大值作为更好的数据分布的上下限？如果没有，有没有办法指定下限和上限，而不是运行min

任何帮助都非常感谢。

Answer 1

对于2000亿行，我希望您的表在您访问数据的同一日期列上在您的数据库中进行分区。没有它，查询将毫无希望。

但是，您是否尝试过日期/时间戳值的整数等效值的下限和上限？查看Spark将整数值转换为时间戳的参考资料。

JDBC选项lowerBound和upperBound转换为TimestampType/DateType值的方式与将字符串转换为TimestampType/DeteType值的方法相同。转换基于Proleptic Gregorian日历和SQL配置spark.SQL.session.timeZone定义的时区。在Spark 2.4及以下版本中，转换基于混合日历（朱利安·格里高利）和默认系统时区。

正如你提到的，没有预先存在的整数列可以在这里使用。所以在你的循环中，上限和下限是静态的，因此可以转换为静态的上限和下限数值。根据Spark的内部结构，下限和上限值被划分为数字范围，并向数据库抛出多个查询，以每次查询获取单个分区的数据。这也意味着在相关列上进行表分区或在源数据库中具有适当的索引对性能非常重要。

您需要确保上限和下限的占位符适当地放置在您提供的查询中。作为提醒；实际数值可能会因使用的数据库系统而异。如果弹出该场景，即数据库系统到日期的整数转换不同，那么您需要提供数据库而不是Spark接受的值。来自相同的文档：

参数：connectionFactory-返回打开的Connection的工厂。RDD负责关闭连接。sql-查询的文本。查询必须包含两个？用于划分结果的参数的占位符。例如

select title, author from books where ? <= id and id <= ?
   

lowerBound - the minimum value of the first placeholder
upperBound - the maximum value of the second placeholder The lower and upper bounds are inclusive.

...

同样，也可以清楚地看到