docs: public static void main(String[] args) throws Exception {
  
  		final ParameterTool params = ParameterTool.fromArgs(args);
  
  		if (!params.has("lineitem") && !params.has("customer") && !params.has("orders")) {
  			System.err.println("  This program expects data from the TPC-H benchmark as input data.");
  			System.err.println("  Due to legal restrictions, we can not ship generated data.");
  			System.out.println("  You can find the TPC-H data generator at http://www.tpc.org/tpch/.");
  			System.out.println("  Usage: TPCHQuery3 --lineitem <path> --customer <path> --orders <path> [--output <path>]");
  			return;
  		}
  
  		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  
  		env.getConfig().setGlobalJobParameters(params);
  
  		// get input data
  		DataSet<Lineitem> lineitems = getLineitemDataSet(env, params.get("lineitem"));
  		DataSet<Customer> customers = getCustomerDataSet(env, params.get("customer"));
  		DataSet<Order> orders = getOrdersDataSet(env, params.get("orders"));
  
  		// Filter market segment "AUTOMOBILE"
  		customers = customers.filter(
  								new FilterFunction<Customer>() {
  									@Override
  									public boolean filter(Customer c) {
  										return c.getMktsegment().equals("AUTOMOBILE");
  									}
  								});
  
  		// Filter all Orders with o_orderdate < 12.03.1995
  		orders = orders.filter(
  							new FilterFunction<Order>() {
  								private final DateFormat format = new SimpleDateFormat("yyyy-MM-dd");
  								private final Date date = format.parse("1995-03-12");
  
  								@Override
  								public boolean filter(Order o) throws ParseException {
  									return format.parse(o.getOrderdate()).before(date);
  								}
  							});
  
  		// Filter all Lineitems with l_shipdate > 12.03.1995
  		lineitems = lineitems.filter(
  								new FilterFunction<Lineitem>() {
  									private final DateFormat format = new SimpleDateFormat("yyyy-MM-dd");
  									private final Date date = format.parse("1995-03-12");
  
  									@Override
  									public boolean filter(Lineitem l) throws ParseException {
  										return format.parse(l.getShipdate()).after(date);
  									}
  								});
  
  		// Join customers with orders and package them into a ShippingPriorityItem
  		DataSet<ShippingPriorityItem> customerWithOrders =
  				customers.join(orders).where(0).equalTo(1)
  							.with(
  								new JoinFunction<Customer, Order, ShippingPriorityItem>() {
  									@Override
  									public ShippingPriorityItem join(Customer c, Order o) {
  										return new ShippingPriorityItem(o.getOrderKey(), 0.0, o.getOrderdate(),
  												o.getShippriority());
  									}
  								});
  
  		// Join the last join result with Lineitems
  		DataSet<ShippingPriorityItem> result =
  				customerWithOrders.join(lineitems).where(0).equalTo(0)
  									.with(
  											new JoinFunction<ShippingPriorityItem, Lineitem, ShippingPriorityItem>() {
  												@Override
  												public ShippingPriorityItem join(ShippingPriorityItem i, Lineitem l) {
  													i.setRevenue(l.getExtendedprice() * (1 - l.getDiscount()));
  													return i;
  												}
  											})
  								// Group by l_orderkey, o_orderdate and o_shippriority and compute revenue sum
  								.groupBy(0, 2, 3)
  								.aggregate(Aggregations.SUM, 1);
  
  		// emit result
  		if (params.has("output")) {
  			result.writeAsCsv(params.get("output"), "\n", "|");
  			// execute program
  			env.execute("TPCH Query 3 Example");
  		} else {
  			System.out.println("Printing result to stdout. Use --output to specify output path.");
  			result.print();
  		}
  
  	}