我正在尝试通过JDBC在单个节点(local [*])上以独立模式访问中型Teradata表(约1亿行)。
我正在使用Spark 1.4.1。并安装在功能非常强大的计算机(2 cpu,24核,126G RAM)上。
我尝试了几种内存设置和调整选项,以使其运行更快,但是它们都没有产生很大的影响。
我确定我缺少一些东西,下面是我的最后一次尝试,它花费了大约11分钟来获得这个简单的计数,而通过R的JDBC连接使用它仅花费了40秒来获得计数。
bin/pyspark --driver-memory 40g --executor-memory 40g
df = sqlContext.read.jdbc("jdbc:teradata://......)
df.count()
当我尝试使用BIG表(5B记录)时,查询完成后未返回任何结果。
最佳答案
将整个数据集检索到内存中的DataFrame
集合后,将执行所有聚合操作。因此,在Spark中进行计数永远不会像直接在TeraData中那样高效。有时值得通过以下方式将一些计算推送到数据库中:创建 View ,然后使用JDBC API映射这些 View 。
每次使用JDBC驱动程序访问大表时,都应指定分区策略,否则将创建具有单个分区的DataFrame
/RDD
,并且将使单个JDBC连接过载。
相反,您想尝试以下AI(自Spark 1.4.0+起):
sqlctx.read.jdbc(
url = "<URL>",
table = "<TABLE>",
columnName = "<INTEGRAL_COLUMN_TO_PARTITION>",
lowerBound = minValue,
upperBound = maxValue,
numPartitions = 20,
connectionProperties = new java.util.Properties()
)
还有一个选项可以降低某些过滤条件。
如果没有统一分布的整数列,则要通过指定自定义谓词(
where
语句)来创建一些自定义分区。例如,假设您有一个timestamp列,并想按日期范围划分: val predicates =
Array(
"2015-06-20" -> "2015-06-30",
"2015-07-01" -> "2015-07-10",
"2015-07-11" -> "2015-07-20",
"2015-07-21" -> "2015-07-31"
)
.map {
case (start, end) =>
s"cast(DAT_TME as date) >= date '$start' AND cast(DAT_TME as date) <= date '$end'"
}
predicates.foreach(println)
// Below is the result of how predicates were formed
//cast(DAT_TME as date) >= date '2015-06-20' AND cast(DAT_TME as date) <= date '2015-06-30'
//cast(DAT_TME as date) >= date '2015-07-01' AND cast(DAT_TME as date) <= date '2015-07-10'
//cast(DAT_TME as date) >= date '2015-07-11' AND cast(DAT_TME as date) <= date //'2015-07-20'
//cast(DAT_TME as date) >= date '2015-07-21' AND cast(DAT_TME as date) <= date '2015-07-31'
sqlctx.read.jdbc(
url = "<URL>",
table = "<TABLE>",
predicates = predicates,
connectionProperties = new java.util.Properties()
)
它将生成一个
DataFrame
,其中每个分区将包含与不同谓词关联的每个子查询的记录。在DataFrameReader.scala检查源代码
关于apache-spark - 如何使用DataFrame和JDBC连接为缓慢的Spark作业提高性能?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/32188295/