INFO:distributed.protocol.pickle:Failed to serialize

[afalamna]

embed = hub.load("https://tfhub.dev/google/universal-sentence-encoder-multilingual/3")

df2 = dd.from_pandas(pd.DataFrame({'ngram': ngrams_list}), npartitions=4)  
csv_dask = dd.read_csv(csv_file_path)

csv_dask = csv_dask.repartition(npartitions=1).reset_index()


def emb_skill(skill):

  return embed([skill])[0].numpy()

df2 = df2.assign(embeddings = df2['ngram'].map_partitions(lambda series: series.apply(emb_skill), meta='object'))
csv_dask = csv_dask.assign(embeddings = csv_dask['hcms_skills'].map_partitions(lambda series: series.apply(emb_skill), meta='object'))

csv_dask.reset_index()

df2.compute()
csv_dask.compute()

This is a snippet of my code I want to use dask distributed to make it faster but it gives me this error whenever I am trying can you please help and tell me what I am doing wrong?

INFO:distributed.protocol.pickle:Failed to serialize (<function map_chunk at 0x7e8804b995a0>, Delayed('emb_skill-4a6ac665-a176-4bea-9f6c-c278abbf6062'), ["('from_sequence-956324e59717b7e4a6b95e21dce2c68b', 0)"], None, {}). Exception: can't pickle repeated message fields, convert to list first
2023-08-09 16:30:40,955 - distributed.protocol.core - CRITICAL - Failed to Serialize

[fjetter]

Can you please share a reproducible example? The code you provided is incomplete.

See https://matthewrocklin.com/minimal-bug-reports for some recommendations about how to minimize your reproducer

[afalamna]

!pip install dask
!pip install "tensorflow-text==2.11.*"
!pip install bokeh
!pip install simpleneighbors[annoy]
!pip install tqdm
!pip3 install tensorflow_text>=2.0.0rc0

import dask.dataframe as dd
import dask.bag as db
import dask
import pandas as pd
import numpy as np
import tensorflow as tf
import tensorflow_hub as hub
import tensorflow_text
from sklearn.metrics.pairwise import cosine_similarity
from nltk.util import ngrams
import dask.delayed as delayed
from dask.distributed import Client, LocalCluster
from itertools import chain
from itertools import product
import re
from dask import delayed


embed = hub.load("https://tfhub.dev/google/universal-sentence-encoder-multilingual/3")


cluster = LocalCluster() 
client = Client(cluster)


# Function to tokenize the text and create n-grams
def create_ngrams(text, n):
    cleaned_text = re.sub(r'[^\w\s]', '', text.lower())
    words = cleaned_text.split()
    ngrams_list = list(ngrams(words, n))
    ngrams_strings = [' '.join(gram) for gram in ngrams_list]
    return ngrams_strings

# Read the text file
with open('', 'r') as file:
    text = file.read()
csv_file_path = ''


# Define the n value for n-grams
n = 3  # Change this value to set the desired n-gram size

ngrams_list = create_ngrams(text, n)

# Create a Dask DataFrame with n-grams as rows
df2 = dd.from_pandas(pd.DataFrame({'ngram': ngrams_list}), npartitions=4)  # Adjust the number of partitions as needed
csv_dask = dd.read_csv(csv_file_path)


csv_dask = csv_dask.repartition(npartitions=1).reset_index()


def emb_skill(skill):
    return embed([skill])[0].numpy()

# Wrap the function using dask.delayed
emb_skill_delayed = delayed(emb_skill)


df2 = df2.assign(embeddings=df2['ngram'].map_partitions(lambda series: series.apply(emb_skill_delayed), meta='object'))
csv_dask = csv_dask.assign(embeddings=csv_dask['hcms_skills'].map_partitions(lambda series: series.apply(emb_skill_delayed), meta='object'))


csv_dask.reset_index()

df2.compute()
csv_dask.compute()

error :
INFO:distributed.protocol.pickle:Failed to serialize subgraph_callable-8647648b-9808-40d2-a20d-0c1d276ca2bc. Exception: can't pickle repeated message fields, convert to list first