New streaming stats implementation

- Uses StatCounter from spark utils
- Added unit test