assembly = Pipes.initial_filters(assembly, start_date, end_date, use_cohort_assign, new_user_only, creation_game_id, game_play_game_id);
        assembly = new Each(assembly, Fields.ALL, new NewBucketFunc(new Fields("cohort", "fb_id", "gsn_event", "number")), new Fields("cohort", "fb_id", "gsn_event", "number"));


        Pipe uniqpipe = new Pipe("uniques", assembly);
        Pipe countpipe = new Pipe("counts",assembly);
        Pipe sumpipe = new Pipe("sums",assembly);

        uniqpipe = new Each(uniqpipe, new Fields("gsn_event"), new ExpressionFilter("!(gsn_event == 1 || gsn_event == 2 || gsn_event == 3 || gsn_event >=19)",
                new String[]{"gsn_event"}, new Class[]{int.class}) );
        uniqpipe = new Unique("uniquefilter", uniqpipe, new Fields("cohort", "gsn_event", "fb_id"), 10000);
        uniqpipe = new Count("uniqct", uniqpipe, new Fields("cohort", "gsn_event"), new Fields("count"), 10000);


        countpipe = new Each(countpipe, new Fields("gsn_event"), new ExpressionFilter("!(gsn_event >= 7 && gsn_event <= 15)",
                new String[]{"gsn_event"}, new Class[]{int.class}) );

        countpipe = new Count("totalct", countpipe, new Fields("cohort", "gsn_event"), new Fields("count"), 10000);

        sumpipe = new Each(sumpipe, new Fields("gsn_event"), new ExpressionFilter("!( (gsn_event >= 16 && gsn_event <= 18) || (gsn_event >= 4 && gsn_event <= 6) )",
                new String[]{"gsn_event"}, new Class[]{int.class}));

        sumpipe = new Sum("totalsum", sumpipe, new Fields("cohort","gsn_event"), new Fields("number"), new Fields("count"), double.class);


        Pipe[] joinpipes = new Pipe[]{uniqpipe, countpipe, sumpipe};


        Pipe joinedresult = new GroupBy("biggroup", joinpipes, new Fields("cohort"));