我创建了一个简单的 Java 应用程序,它使用 Apache Spark 从 Cassandra 检索数据,对其进行一些转换并将其保存在另一个 Cassandra 表中。
我正在使用 Apache Spark 1.4.1,它配置为独立集群模式,具有单个主服务器和从服务器,位于我的计算机上。
DataFrame customers = sqlContext.cassandraSql("SELECT email, first_name, last_name FROM customer " +
"WHERE CAST(store_id as string) = '" + storeId + "'");
DataFrame customersWhoOrderedTheProduct = sqlContext.cassandraSql("SELECT email FROM customer_bought_product " +
"WHERE CAST(store_id as string) = '" + storeId + "' AND product_id = " + productId + "");
// We need only the customers who did not order the product
// We cache the DataFrame because we use it twice.
DataFrame customersWhoHaventOrderedTheProduct = customers
.join(customersWhoOrderedTheProduct
.select(customersWhoOrderedTheProduct.col("email")), customers.col("email").equalTo(customersWhoOrderedTheProduct.col("email")), "leftouter")
.where(customersWhoOrderedTheProduct.col("email").isNull())
.drop(customersWhoOrderedTheProduct.col("email"))
.cache();
int numberOfCustomers = (int) customersWhoHaventOrderedTheProduct.count();
Date reportTime = new Date();
// Prepare the Broadcast values. They are used in the map below.
Broadcast<String> bStoreId = sparkContext.broadcast(storeId, classTag(String.class));
Broadcast<String> bReportName = sparkContext.broadcast(MessageBrokerQueue.report_did_not_buy_product.toString(), classTag(String.class));
Broadcast<java.sql.Timestamp> bReportTime = sparkContext.broadcast(new java.sql.Timestamp(reportTime.getTime()), classTag(java.sql.Timestamp.class));
Broadcast<Integer> bNumberOfCustomers = sparkContext.broadcast(numberOfCustomers, classTag(Integer.class));
// Map the customers to a custom class, thus adding new properties.
DataFrame storeCustomerReport = sqlContext.createDataFrame(customersWhoHaventOrderedTheProduct.toJavaRDD()
.map(row -> new StoreCustomerReport(bStoreId.value(), bReportName.getValue(), bReportTime.getValue(), bNumberOfCustomers.getValue(), row.getString(0), row.getString(1), row.getString(2))), StoreCustomerReport.class);
// Save the DataFrame to cassandra
storeCustomerReport.write().mode(SaveMode.Append)
.option("keyspace", "my_keyspace")
.option("table", "my_report")
.format("org.apache.spark.sql.cassandra")
.save();
正如你所看到的,我cache
the customersWhoHaventOrderedTheProduct
DataFrame,之后我执行count
并打电话toJavaRDD
.
By my calculations these actions should be executed only once. But when I go in the Spark UI for the current job I see the following stages:
正如您所看到的,每个动作都会执行两次。
难道我做错了什么?有什么设置我错过了吗?
任何想法都将不胜感激。
EDIT:
我打电话后System.out.println(storeCustomerReport.toJavaRDD().toDebugString());
这是调试字符串:
(200) MapPartitionsRDD[43] at toJavaRDD at DidNotBuyProductReport.java:93 []
| MapPartitionsRDD[42] at createDataFrame at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[41] at map at DidNotBuyProductReport.java:90 []
| MapPartitionsRDD[40] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[39] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[38] at toJavaRDD at DidNotBuyProductReport.java:89 []
| ZippedPartitionsRDD2[37] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[31] at toJavaRDD at DidNotBuyProductReport.java:89 []
| ShuffledRDD[30] at toJavaRDD at DidNotBuyProductReport.java:89 []
+-(2) MapPartitionsRDD[29] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[28] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[27] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[3] at cache at DidNotBuyProductReport.java:76 []
| CassandraTableScanRDD[2] at RDD at CassandraRDD.scala:15 []
| MapPartitionsRDD[36] at toJavaRDD at DidNotBuyProductReport.java:89 []
| ShuffledRDD[35] at toJavaRDD at DidNotBuyProductReport.java:89 []
+-(2) MapPartitionsRDD[34] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[33] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[32] at toJavaRDD at DidNotBuyProductReport.java:89 []
| MapPartitionsRDD[5] at cache at DidNotBuyProductReport.java:76 []
| CassandraTableScanRDD[4] at RDD at CassandraRDD.scala:15 []
EDIT 2:
因此,经过一些研究并结合试验和错误,我设法优化了这项工作。
我创建了一个 RDDcustomersWhoHaventOrderedTheProduct
我在调用之前缓存它count()
行动。 (我将缓存从DataFrame
to the RDD
).
之后我用这个RDD
来创建storeCustomerReport
DataFrame
.
JavaRDD<Row> customersWhoHaventOrderedTheProductRdd = customersWhoHaventOrderedTheProduct.javaRDD().cache();
现在各个阶段如下所示:
正如你所看到的两个count
and cache
现在已经消失了,但仍然有两个“javaRDD”操作。我不知道他们从哪里来,正如我所说的toJavaRDD
我的代码中只出现过一次。