Rewritten more to graphstage

I think we should rewrite more stream operators to graphstages, there are some are not.