Apache Spark を使用して Cassandra からデータを取得し、データを変換して別の Cassandra テーブルに保存する単純な Java アプリケーションを作成しました。
マシン上にある単一のマスターとスレーブを使用して、スタンドアロン クラスター モードで構成された 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のcustomersWhoHaventOrderedTheProductとおり、DataFrame を実行した後、 を実行してcountを呼び出しますtoJavaRDD。
私の計算では、これらのアクションは 1 回だけ実行する必要があります。しかし、現在のジョブの Spark UI に移動すると、次のステージが表示されます。
ご覧のとおり、すべてのアクションが 2 回実行されます。
私は何か間違ったことをしていますか？見逃した設定はありますか？
どんなアイデアでも大歓迎です。
編集：
私が電話した後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 []
編集2：
そのため、いくつかの調査と試行錯誤を組み合わせた結果、仕事を最適化することができました。
から RDD を作成し、アクションcustomersWhoHaventOrderedTheProductを呼び出す前にそれをキャッシュします。count()(キャッシュを から に移動しDataFrameましたRDD)。
その後、これRDDを使用してstoreCustomerReport DataFrame.
JavaRDD<Row> customersWhoHaventOrderedTheProductRdd = customersWhoHaventOrderedTheProduct.javaRDD().cache();
ステージは次のようになります。
ご覧のとおり、2 つのcountとcacheはなくなりましたが、まだ 2 つの「javaRDD」アクションがあります。toJavaRDDコード内で 1 回しか呼び出していないため、それらがどこから来ているのかわかりません。