Spark DataFrame-最后一个分区收集慢速

Question

我有一个Java片段，该摘要从远程Oracle DB（至少65K记录）中读取记录。本质上，我们正在尝试将每小时过滤器传递到DataFrame，以获取记录，以每小时分区x 24。

。

源视图基于具有数百万记录的表。

我们面临的问题是，在不到3分钟的24个分区中，火花（在纱线或火花集群上）处理22分。最后两个分区需要超过5个小时才能完成。

有什么办法可以使用DataFrames加速此操作？

HashMap<String, String> options = new HashMap<>();
sqlContext.setConf("spark.sql.shuffle.partition", "50");
options.put("dbtable", "( select * from "+VIEW_NAME+" where 1=1)");
options.put("driver", "oracle.jdbc.OracleDriver");
options.put("url", JDBC_URL);
options.put("partitionColumn", "hrs");
options.put("lowerBound", "00");
options.put("upperBound", "23");
options.put("numPartitions", "24");
DataFrame dk = sqlContext.load("jdbc", options).cache();   
dk.registerTempTable(VIEW_NAME);
dk.printSchema();
DateTime dt = new DateTime(2015, 5, 8, 10, 0, 0);
String s = SQL_DATE_FORMATTER.print(dt);
dt = dt.plusHours(24);
String t = SQL_DATE_FORMATTER.print(dt);
System.out.println("S is " + s + "and t is "+ t);
Stream<Row> rows = dk.filter("DATETIME >= '" + s + "' and DATETIME <= '" + t + "'").collectAsList().parallelStream();
    System.out.println("Collected" + rows.count());

Answer 1

不确定这是否是完整的答案，但是作为工作，如果我们执行以下

dt = dt.plusHours(24).minusSeconds(1)

它更快，但仍然不如第一个23个分区