is there any way to do this directly.  if its not, is there any todo this indirectly using another datastrcutures of spark