ValueError: When input_signature is provided, all inputs to the Python function must be convertible to tensors:

user_10994166 0
Viewed 109

I have pyspark df like this:

id        desc
1      abd hdbh jbj
2      sgjhd jhdgh gjhg
3      bvj hvhgvgh
4      jkjb bhj

Now I want to convert my desc column to vector so I'm using Google sentence encoder as udf, here's my code:

module_url = "https://tfhub.dev/google/universal-sentence-encoder/4"
model = hub.load(module_url)

def embedding(input):
    return (model[input])

df.withColumn("Embedding", list(embedding(f.lit("desc"))))

Here's the error log:

ValueError                                Traceback (most recent call last)
/tmp/ipykernel_13810/1173342766.py in <module>
----> 1 df_shirt_sample.withColumn("Embedding", list(embedding(f.lit("desc"))))

/tmp/ipykernel_13810/446837446.py in embedding(input)
      1 def embedding(input):
----> 2     return (model(input))

~/miniconda3/envs/dev_env_37/lib/python3.7/site-packages/tensorflow/python/saved_model/load.py in _call_attribute(instance, *args, **kwargs)
    684 
    685 def _call_attribute(instance, *args, **kwargs):
--> 686   return instance.__call__(*args, **kwargs)
    687 
    688 

~/miniconda3/envs/dev_env_37/lib/python3.7/site-packages/tensorflow/python/util/traceback_utils.py in error_handler(*args, **kwargs)
    151     except Exception as e:
    152       filtered_tb = _process_traceback_frames(e.__traceback__)
--> 153       raise e.with_traceback(filtered_tb) from None
    154     finally:
    155       del filtered_tb

~/miniconda3/envs/dev_env_37/lib/python3.7/site-packages/tensorflow/python/eager/function_spec.py in _convert_inputs_to_signature(inputs, input_signature, flat_input_signature)
    521         need_packing = True
    522       except ValueError:
--> 523         raise ValueError("When input_signature is provided, all inputs to "
    524                          "the Python function must be convertible to "
    525                          "tensors:\n"

ValueError: When input_signature is provided, all inputs to the Python function must be convertible to tensors:
  inputs: (
    Column<b'desc'>)
  input_signature: (
    TensorSpec(shape=<unknown>, dtype=tf.string, name=None)).

Can someone tell me what I'm doing wrong

tensorflow-hub word2vec pyspark tensorflow python
1 Answers

I've tried to load the model from a UDF following this post: Spark broadcast a trained tensorflow SavedModel. This will load a model on each worker and then you can make the predictions.

Example:

import tensorflow_hub as hub
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, FloatType


def embedding(x):
    model_url = "https://tfhub.dev/google/universal-sentence-encoder/4"
    model = hub.load(model_url, tags=["serve"])
    return model([x])


@F.udf(returnType=ArrayType(ArrayType(FloatType())))
def infer(data):
    outputs = embedding(data)
    return outputs.numpy().tolist()


spark = SparkSession.builder.getOrCreate()
data = [{"text": "abd hdbh jbj"}]
df = spark.createDataFrame(data=data)
df = df.withColumn("embedding", infer("text"))
df.show(10)
df.printSchema()

Which gives:

+------------+--------------------+                                             
|        text|           embedding|
+------------+--------------------+
|abd hdbh jbj|[[0.054931916, -0...|
+------------+--------------------+

root
 |-- text: string (nullable = true)
 |-- embedding: array (nullable = true)
 |    |-- element: array (containsNull = true)
 |    |    |-- element: float (containsNull = true)

Check out the Guide to Deploying a TF 2.0 SavedModel on Spark using PySpark: https://github.com/tensorflow/tensorflow/issues/31421

Another way to do this is to use Petastorm to convert the Spark Dataframe to TensorFlow Dataframe and then feed it to a distributed model, please see: https://www.databricks.com/notebooks/simple-aws/petastorm-spark-converter-tensorflow.html

user_109479970
Related