Replace map.flatten(1) with flat_map