I am having an incredibly bad time working with document vectors produced by the Spacy pre-trained large model in a Pyspark environment on AWS. The problems specifically start when I put the document vectors into a dataframe.
For example, this code works just fine for me:
# Load infrastructure libraries
import pandas as pd
import numpy as np
# Load NLP libraries and tools
import spacy
# Prepare the Spacy NLP parsers
nlp = spacy.load('en_core_web_lg')
# Load Spark
from pyspark.sql.session import SparkSession
import pyspark.sql.functions as F
from pyspark.sql import types as T
from pyspark.ml.linalg import Vectors
from pyspark.ml.functions import vector_to_array
# Setup spark sesssion
spark = SparkSession.builder.appName("Spacy Test").getOrCreate()
# Create a test document vector
vec = nlp("I really like cheese").vector
vec = Vectors.dense(vec)
vec
And the output:
DenseVector([-0.7719, -0.152, -2.5687, -2.9057, -4.3302, -0.2236, 1.4182, 4.7625, -5.4947, 4.0054, 5.4234, 0.4195, -2.1278, -0.2198, 3.733, -4.3451, 1.6354, -4.2959, -1.9893, -1.0955, 1.2923, 2.5803, 0.8446, -5.6266, -2.1173, -3.1521, -3.2044, 1.3603, -1.8462, 0.6012, -0.0218, -3.5514, 0.0648, 1.1473, 0.8563, -3.0121, -1.114, 0.7547, 3.3102, 2.2526, -1.9224, 0.9549, 0.4257, -3.7478, 3.382, 2.2748, -2.0761, -4.0411, -2.0196, 2.7873, 1.1951, 0.0311, 1.0897, -3.7514, -2.9269, -0.8296, 0.446, 3.5525, 3.3304, 1.706, 6.1961, -1.7517, 0.5205, -0.1543, -2.7567, -0.3654, -4.5379, -3.7601, 0.6534, 3.3904, 0.0462, 2.9809, 2.0856, -0.6889, 5.0641, 1.9436, -2.7128, -1.2188, -1.1388, -4.5075, -3.0456, 1.5704, 6.1288, 0.1904, 1.4656, 0.0181, 1.4148, -1.4573, -0.9094, -1.7613, -2.7317, 0.4354, 2.9197, -6.9938, -0.6905, -3.1972, 2.4815, -4.285, 1.0154, -2.5749, 0.8907, 1.9208, -0.1247, 1.3449, -4.1218, 2.466, -3.8912, 1.6637, -2.4908, -0.7045, -0.8058, -3.7729, 3.0547, -1.9908, -0.085, 1.4266, -0.9306, -2.9857, 0.2251, -0.5722, -2.8197, -0.6044, -4.2039, 2.7789, -0.9341, -0.7502, 1.7431, -3.3388, 0.3382, -0.6125, 0.0842, 1.545, 1.432, -1.2881, -0.123, 0.8724, -4.9354, -2.8043, 5.0844, -2.8918, -2.777, -0.0504, 0.5404, -0.1609, -6.5542, -0.9069, 0.1638, 0.4534, -3.2973, -3.5313, -2.8212, 0.9449, -3.5691, 0.2708, -2.7588, 2.1502, 6.3973, 1.1362, -1.0846, -1.0468, -0.6148, -3.8309, -3.2811, 0.1241, -0.4777, 0.6669, -0.9383, -0.869, -2.0688, 2.1084, 2.3942, -0.8358, -0.3179, 3.1491, 1.8569, -3.479, -2.1367, -0.7273, 1.0882, -0.9835, -1.2419, 0.5295, 5.0464, -1.2607, -2.0058, 2.4997, 0.7322, -5.4079, 2.3466, 2.298, -2.1624, 4.132, -2.6485, 2.1919, -1.6362, -2.5953, -5.571, 0.4838, -1.8915, 5.4007, -1.4195, -0.3522, -3.0002, -1.0116, -0.3323, 2.2401, 2.1777, -0.616, -0.0131, -0.2913, 1.5619, -1.0237, 1.9203, 1.3495, -3.4587, 0.6858, -1.129, 0.58, -1.5449, -0.3716, -1.0494, -1.689, -1.4535, 4.1873, 2.1134, -2.9858, -1.2115, 1.3913, 3.0331, 2.1075, -7.7124, -2.9558, 0.8318, -1.1737, -1.575, 0.6175, -2.9589, 6.8081, 0.4126, -1.2054, -5.8529, -1.4182, 1.9067, 2.4857, 1.5191, -2.5983, 0.8784, -0.2694, 3.1889, 0.6379, -3.4795, 4.4023, -7.337, 0.9995, -0.8919, -6.4528, -1.2682, 2.664, -3.828, 3.5323, 3.0402, 3.2275, 1.1329, 2.3996, 2.9425, -0.3751, 1.7857, 1.2017, -6.3812, 1.7977, 0.4254, -2.3142, -2.666, 1.8858, -0.5762, -2.9764, 2.7129, -2.1113, 0.3109, 1.2368, -4.8247, -5.0767, 1.0253, 0.2394, 3.9423, -1.2326, 0.3167, -0.0368, -3.8825, 2.8392, 2.399, -1.3454, 2.2132, -2.8337, -2.822, -0.3926, 3.3747, 5.8344, -4.1289, 2.2593])
However, my next step is to derive the document vectors for a target field in a dataframe. This target field contains a series of documents, with each document stored as a string, one document per row. Here's an analogous example:
data = [
("1", "I really like cheese", 0.35),
("1", "I don't really like cheese", 0.10),
("1", "I absolutely love cheese", 0.55)
]
from pyspark.sql.types import FloatType, ArrayType, StringType, StructType, StructField
schema = StructType([
StructField("id",StringType(),True),
StructField("target",StringType(),True),
StructField("pct",FloatType(),True),
])
df = spark.createDataFrame(data=data,schema=schema)
df.show()
root
|-- id: string (nullable = true)
|-- target: string (nullable = true)
|-- pct: float (nullable = true)
+---+--------------------+----+
| id| target| pct|
+---+--------------------+----+
| 1|I really like cheese|0.35|
| 1|I don't really li...| 0.1|
| 1|I absolutely love...|0.55|
+---+--------------------+----+
The "pct" column in my dataframe is there because I eventually want to multiply each embedding vector by this column, but we don't need to worry about that right now. All of the code up to this point works just fine.
My next step is to get the embedding vectors for each document:
embedding_udf = F.udf(lambda x: get_embeddings(x))
def get_embeddings(x):
# Return
return nlp(x).vector
toDense_udf = F.udf(lambda v: toDense(v))
def toDense(v):
v = Vectors.dense(v)
return v
# Get the embedding vectors for each document in the dataframe
df = df.withColumn("embedding", embedding_udf(F.col("target"))).select(
"id",
"embedding",
"pct"
)
# Convert the embeddings to dense vectors
df = df.withColumn("embedding", toDense_udf(F.col("embedding")))
And this is where I start to have problems. Now, this code runs just fine, but I cannot do any kind of analysis or debugging because every attempt to examine this dataframe after the vectors are generating is failing.
Running:
df.show()
Throws this error:
An error was encountered:
An error occurred while calling o127.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 4.0 failed 4 times, most recent failure: Lost task 0.3 in stage 4.0 (TID 7) (ip-10-0-0-162.ec2.internal executor 9): java.lang.RuntimeException: Failed to run command: /usr/bin/virtualenv -p python3 --system-site-packages virtualenv_application_1663597828866_0004_0
at org.apache.spark.api.python.VirtualEnvFactory.execCommand(VirtualEnvFactory.scala:120)
at org.apache.spark.api.python.VirtualEnvFactory.setupVirtualEnv(VirtualEnvFactory.scala:78)
at org.apache.spark.api.python.PythonWorkerFactory.<init>(PythonWorkerFactory.scala:94)
at org.apache.spark.SparkEnv.$anonfun$createPythonWorker$1(SparkEnv.scala:125)
at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:86)
at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:125)
at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:162)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec.evaluate(BatchEvalPythonExec.scala:81)
at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$2(EvalPythonExec.scala:130)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:863)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:863)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:133)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1474)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2610)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2559)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2558)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2558)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1200)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1200)
at scala.Option.foreach(Option.scala:407)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1200)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2798)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2740)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2729)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:978)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2215)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2236)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2255)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:519)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:472)
at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:48)
at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3779)
at org.apache.spark.sql.Dataset.$anonfun$head$1(Dataset.scala:2769)
at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3770)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:107)
at org.apache.spark.sql.execution.SQLExecution$.withTracker(SQLExecution.scala:232)
at org.apache.spark.sql.execution.SQLExecution$.executeQuery$1(SQLExecution.scala:110)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:135)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:107)
at org.apache.spark.sql.execution.SQLExecution$.withTracker(SQLExecution.scala:232)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:135)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:253)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:134)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:68)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3768)
at org.apache.spark.sql.Dataset.head(Dataset.scala:2769)
at org.apache.spark.sql.Dataset.take(Dataset.scala:2976)
at org.apache.spark.sql.Dataset.getRows(Dataset.scala:289)
at org.apache.spark.sql.Dataset.showString(Dataset.scala:328)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.RuntimeException: Failed to run command: /usr/bin/virtualenv -p python3 --system-site-packages virtualenv_application_1663597828866_0004_0
at org.apache.spark.api.python.VirtualEnvFactory.execCommand(VirtualEnvFactory.scala:120)
at org.apache.spark.api.python.VirtualEnvFactory.setupVirtualEnv(VirtualEnvFactory.scala:78)
at org.apache.spark.api.python.PythonWorkerFactory.<init>(PythonWorkerFactory.scala:94)
at org.apache.spark.SparkEnv.$anonfun$createPythonWorker$1(SparkEnv.scala:125)
at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:86)
at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:125)
at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:162)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec.evaluate(BatchEvalPythonExec.scala:81)
at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$2(EvalPythonExec.scala:130)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:863)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:863)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:133)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1474)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
Traceback (most recent call last):
File "/mnt1/yarn/usercache/livy/appcache/application_1663597828866_0004/container_1663597828866_0004_01_000001/pyspark.zip/pyspark/sql/dataframe.py", line 495, in show
print(self._jdf.showString(n, 20, vertical))
File "/mnt1/yarn/usercache/livy/appcache/application_1663597828866_0004/container_1663597828866_0004_01_000001/py4j-0.10.9.3-src.zip/py4j/java_gateway.py", line 1322, in __call__
answer, self.gateway_client, self.target_id, self.name)
File "/mnt1/yarn/usercache/livy/appcache/application_1663597828866_0004/container_1663597828866_0004_01_000001/pyspark.zip/pyspark/sql/utils.py", line 111, in deco
return f(*a, **kw)
File "/mnt1/yarn/usercache/livy/appcache/application_1663597828866_0004/container_1663597828866_0004_01_000001/py4j-0.10.9.3-src.zip/py4j/protocol.py", line 328, in get_return_value
format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o127.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 4.0 failed 4 times, most recent failure: Lost task 0.3 in stage 4.0 (TID 7) (ip-10-0-0-162.ec2.internal executor 9): java.lang.RuntimeException: Failed to run command: /usr/bin/virtualenv -p python3 --system-site-packages virtualenv_application_1663597828866_0004_0
at org.apache.spark.api.python.VirtualEnvFactory.execCommand(VirtualEnvFactory.scala:120)
at org.apache.spark.api.python.VirtualEnvFactory.setupVirtualEnv(VirtualEnvFactory.scala:78)
at org.apache.spark.api.python.PythonWorkerFactory.<init>(PythonWorkerFactory.scala:94)
at org.apache.spark.SparkEnv.$anonfun$createPythonWorker$1(SparkEnv.scala:125)
at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:86)
at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:125)
at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:162)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec.evaluate(BatchEvalPythonExec.scala:81)
at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$2(EvalPythonExec.scala:130)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:863)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:863)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:133)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1474)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2610)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2559)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2558)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2558)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1200)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1200)
at scala.Option.foreach(Option.scala:407)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1200)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2798)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2740)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2729)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:978)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2215)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2236)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2255)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:519)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:472)
at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:48)
at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3779)
at org.apache.spark.sql.Dataset.$anonfun$head$1(Dataset.scala:2769)
at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3770)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:107)
at org.apache.spark.sql.execution.SQLExecution$.withTracker(SQLExecution.scala:232)
at org.apache.spark.sql.execution.SQLExecution$.executeQuery$1(SQLExecution.scala:110)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:135)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:107)
at org.apache.spark.sql.execution.SQLExecution$.withTracker(SQLExecution.scala:232)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:135)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:253)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:134)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:68)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3768)
at org.apache.spark.sql.Dataset.head(Dataset.scala:2769)
at org.apache.spark.sql.Dataset.take(Dataset.scala:2976)
at org.apache.spark.sql.Dataset.getRows(Dataset.scala:289)
at org.apache.spark.sql.Dataset.showString(Dataset.scala:328)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.RuntimeException: Failed to run command: /usr/bin/virtualenv -p python3 --system-site-packages virtualenv_application_1663597828866_0004_0
at org.apache.spark.api.python.VirtualEnvFactory.execCommand(VirtualEnvFactory.scala:120)
at org.apache.spark.api.python.VirtualEnvFactory.setupVirtualEnv(VirtualEnvFactory.scala:78)
at org.apache.spark.api.python.PythonWorkerFactory.<init>(PythonWorkerFactory.scala:94)
at org.apache.spark.SparkEnv.$anonfun$createPythonWorker$1(SparkEnv.scala:125)
at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:86)
at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:125)
at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:162)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec.evaluate(BatchEvalPythonExec.scala:81)
at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$2(EvalPythonExec.scala:130)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:863)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:863)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:133)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1474)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
In fact, every attempt I make to examine any of individual vectors in the dataframe fail with a very similar error. I've tried using collect(), I've tried converting the dataframe to a pandas dataframe, but everything fails with some kind of verbose error. So my first question is, Why does every attempt to view the embedding vectors in a dataframe fail? I can't even convert this basic 3-row dataframe to a pandas dataframe without errors. What am I doing wrong?
And secondly, the data type of the field containing the vectors is throwing me off because the embedding field is apparently being stored as a string? For example,
Running:
df.dtypes
Yields:
[('id', 'string'), ('embedding', 'string'), ('pct', 'float')]
So my second question is, why does the column containing the document vectors have a string datatype? This doesn't seem right, especially since the data type of an individual string vector is of type numpy.ndarray:
type(nlp("I really like cheese").vector)
<class 'numpy.ndarray'>
From what I can tell, there is virtually no documentation for integrating Spacy with Pyspark in this way, even though all of these operations should be quite simple in a regular python environment. Any insights would be greatly appreciated.
My environment details:
Release label:emr-6.7.0
Hadoop distribution:Amazon 3.2.1
Applications:Spark 3.2.1, Livy 0.7.1, JupyterEnterpriseGateway 2.1.0
Thanks!