spark-dataframes-getting-started.java

package spark2demo;

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.spark.sql.functions.sum;
import static org.apache.spark.sql.functions.round;

public class GettingStarted {
	public static void main(String[] args) {														// your system
		SparkSession spark = SparkSession.
				builder().
				appName("Simple Application").
				master("local").
				getOrCreate();
		
		String ordersPath = "/Users/itversity/Research/data/retail_db/orders";
		Dataset<Row> orders = spark.
				read().
				schema("order_id INT, order_date STRING, order_customer_id INT, order_status STRING").
				format("csv").
				load(ordersPath).
				cache();

		Dataset<Row> ordersFiltered = orders.
				filter("order_status IN ('COMPLETE', 'CLOSED')");
		
		String orderItemsPath = "/Users/itversity/Research/data/retail_db/order_items";
		Dataset<Row> orderItems = spark.
				read().
				schema("order_item_id INT, order_item_order_id INT, " 
						+ "order_item_product_id INT, order_item_quantity INT, "
						+ "order_item_subtotal FLOAT, order_item_product_price FLOAT").
				format("csv").
				load(orderItemsPath);
		
		Dataset<Row> ordersJoin = ordersFiltered.
				join(orderItems, ordersFiltered.col("order_id").equalTo(orderItems.col("order_item_order_id")));
		
		ordersJoin.
			groupBy("order_date").
			agg(round(sum(ordersJoin.col("order_item_subtotal")), 2).alias("revenue")).
			show();
		
		spark.stop();
	}
}