Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix metadata inference with pandas and dask #35

Merged
merged 3 commits into from
Apr 19, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion nemo_curator/modules/filter.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,19 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import pandas as pd
from dask.dataframe.extensions import make_array_nonempty
from dask.typing import no_default

from nemo_curator.datasets import DocumentDataset
from nemo_curator.utils.module_utils import is_batched

# Override so that pd.NA is not passed during the metadata inference
make_array_nonempty.register(
pd.StringDtype,
lambda x: pd.array(["a", "b"], dtype=x),
)


class Score:
def __init__(self, score_fn, score_field, text_field="text", score_type=None):
Expand Down
12 changes: 11 additions & 1 deletion nemo_curator/modules/task.py
Original file line number Diff line number Diff line change
Expand Up @@ -302,6 +302,8 @@ def _threshold_ngram_count(self, matched_ngrams: dict) -> set:
return filtered_ngrams

def _remove_ngrams_partition(self, partition, task_ngrams, ngrams_freq_sorted):
text_type = partition[self.text_field].dtype

document_fn = partial(
self._remove_ngrams,
task_ngrams=task_ngrams,
Expand All @@ -318,7 +320,15 @@ def _remove_ngrams_partition(self, partition, task_ngrams, ngrams_freq_sorted):

partition[self.text_field] = split_text
filtered_partition = partition[valid_documents_mask]
return filtered_partition.explode(self.text_field, ignore_index=True)
exploded_partition = filtered_partition.explode(
self.text_field, ignore_index=True
)
# After exploding, the string datatype can become an "object" type
exploded_partition[self.text_field] = exploded_partition[
self.text_field
].astype(text_type)

return exploded_partition
ryantwolf marked this conversation as resolved.
Show resolved Hide resolved

def _remove_ngrams(self, document, task_ngrams, ngrams_freq_sorted):
"""
Expand Down