Spark: [BUG]: VectorUDF error

Created on 3 Dec 2019  路  13Comments  路  Source: dotnet/spark

Hi,

i have the following code running using
Spark 2.4.4, Microsoft.Spark 0.6.0 on windows box.

      var schemaToReturn = new StructType(new[]
              {
                    new StructField("codeToGroup", new StringType()),
                    new StructField("count", new IntegerType())
              });

      var mixedDF1 = filtered.GroupBy("codeToGroup")
                              .Apply(schemaToReturn, batchRecord =>Count(batchRecord, "codeToGroup", "name"));

    private static RecordBatch Count(RecordBatch records, string groupFieldName, string nameField)
    {
          int returnLength = records.Length > 0 ? 1 : 0;
            int groupFieldIndex = records.Schema.GetFieldIndex(groupFieldName);
            Field groupField = records.Schema.GetFieldByIndex(groupFieldIndex);

            int nameFieldIndex = records.Schema.GetFieldIndex(nameField);

            StringArray names = records.Column(nameFieldIndex) as StringArray;

            int currentCount = 0;

            for (int i = 0; i < names.Length; ++i)
            {
                var current = names.GetString(i);
                currentCount += (current?.Length ?? 0);
            }

            var outputSchema = new Schema.Builder()
                                          .Field(groupField)
                                          .Field(f => f.Name("Count").DataType(Int32Type.Default))
                                          .Build();

            var outputData = new IArrowArray[]
                                  {
                                    records.Column(groupFieldIndex),
                                    new Int32Array.Builder().Append(currentCount).Build(),
                                  };
            return new RecordBatch(outputSchema, outputData, returnLength);
    }

This throws the following exception :

[Error] [JvmBridge] JVM method execution failed: Nonstatic method showString failed for class 55 when called with 3 arguments ([Index=1, Type=Int32, Value=20], [Index=2, Type=Int32, Value=150], [Index=3, Type=Boolean, Value=False], )
[Error] [JvmBridge] org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 42.0 failed 1 times, most recent failure: Lost task 0.0 in stage 42.0 (TID 817, localhost, executor driver): java.lang.IllegalArgumentException
at java.nio.ByteBuffer.allocate(ByteBuffer.java:334)
at org.apache.arrow.vector.ipc.message.MessageSerializer.readMessage(MessageSerializer.java:543)
at org.apache.arrow.vector.ipc.message.MessageChannelReader.readNext(MessageChannelReader.java:58)
at org.apache.arrow.vector.ipc.ArrowStreamReader.readSchema(ArrowStreamReader.java:132)
at org.apache.arrow.vector.ipc.ArrowReader.initialize(ArrowReader.java:181)
at org.apache.arrow.vector.ipc.ArrowReader.ensureInitialized(ArrowReader.java:172)
at org.apache.arrow.vector.ipc.ArrowReader.getVectorSchemaRoot(ArrowReader.java:65)
at org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:162)
at org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:122)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:410)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:123)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
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:748)

Any idea what is wrong here ?

bug
Source
picture bolcman

All 13 comments

@elvaliuliuliu / @suhsteve can you help?

imback82 picture imback82 on 5 Dec 2019
👍1

@bolcman your sample code seems very similar to the VectorUdfs examples and I didn't notice anything out of the ordinary with it. From the exception message it's running into issues when a DataFrame.Show() statement is run, which I don't see in your sample code. Can you paste your full code sample ?

suhsteve picture suhsteve on 5 Dec 2019

Sure here is the rest, i forgot to mention, i'm using memsql connector in this case.

        var spark = SparkSession.Builder()
                                 .AppName("Hello Spark!")
                                 .Config("spark.memsql.host", "Host")
                                 .Config("spark.memsql.user", "User")
                                 .Config("spark.memsql.password", "pass")
                                 .GetOrCreate();




        var df = spark.Read().Format("com.memsql.spark.connector")
                             .Option("query", "select code, date, holidayType, 'test' as name from spark_calendar limit 1000").Load();




        df.Show(10);

        var filtered = df.WithColumnRenamed("code", "codeToGroup");


        var schemaToReturn = new StructType(new[]
          {
                new StructField("codeToGroup", new StringType()),
                new StructField("count", new IntegerType())
          });

        var mixedDF1 = filtered.GroupBy("codeToGroup")
                                .Apply(schemaToReturn, batchRecord => Count(batchRecord, "codeToGroup", "name"));

        mixedDF1.PrintSchema();
        mixedDF1.Show(10, 10);

error

bolcman picture bolcman on 6 Dec 2019

@bolcman do you also run into issues when you try to run a modified version of your code against people.json ?

suhsteve picture suhsteve on 9 Dec 2019

@bolcman can you try without selecting the date column?

imback82 picture imback82 on 9 Dec 2019

Without date column it works! Going fwd can we have this fixed ?

bolcman picture bolcman on 9 Dec 2019

@eerhardt may have already fixed this issue in Arrow 0.15.x: https://github.com/apache/arrow/commit/0475455bc7ba4297214e618b77ed4a12e1394414

We will update the Arrow version in the next release. Thanks @bolcman for reporting!

imback82 picture imback82 on 9 Dec 2019
👍1

Can I ask how are you using Apply() after GroupBy() as this is an internal function?

Plasma picture Plasma on 10 Dec 2019

Can you install this nuget: https://www.nuget.org/packages/Microsoft.Spark.experimental?

imback82 picture imback82 on 10 Dec 2019

Can you install this nuget: https://www.nuget.org/packages/Microsoft.Spark.experimental?

Thank you, that's let me access Apply().

Plasma picture Plasma on 10 Dec 2019

Closing this and will follow up with #370.

imback82 picture imback82 on 10 Dec 2019

Need Help.
JVM method execution failed: Nonstatic method 'showString' failed for class '72' when called with 3 arguments
spark.Udf().Register(
"MyUDF",
(name) => name + " My New Name " );
dfEmployee.CreateOrReplaceTempView("dateView");
spark.Sql("Select * From dateView").Show();
DataFrame dfv = spark.Sql("Select MyUDF(Name) FROM dateView");
dfv.Show(10, 10, false);
Getting following error

System.Exception
HResult=0x80131500
Message=JVM method execution failed: Nonstatic method 'showString' failed for class '26' when called with 3 arguments ([Index=1, Type=Int32, Value=20], [Index=2, Type=Int32, Value=20], [Index=3, Type=Boolean, Value=False], )
Source=Microsoft.Spark
StackTrace:
at Microsoft.Spark.Interop.Ipc.JvmBridge.CallJavaMethod(Boolean isStatic, Object classNameOrJvmObjectReference, String methodName, Object[] args)
at Microsoft.Spark.Interop.Ipc.JvmBridge.CallNonStaticJavaMethod(JvmObjectReference objectId, String methodName, Object[] args)
at Microsoft.Spark.Interop.Ipc.JvmObjectReference.Invoke(String methodName, Object[] args)
at Microsoft.Spark.Sql.DataFrame.Show(Int32 numRows, Int32 truncate, Boolean vertical)
at MySparkApp.Program.Main(String[] args) in C:\Users\Administrator\source\repos\WM.POC.Databricks\MySparkApp\Program.cs:line 79

Inner Exception 1:
JvmException: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0 (TID 1, host.docker.internal, executor driver): java.io.IOException: Cannot run program "Microsoft.Spark.Worker.exe": CreateProcess error=2, The system cannot find the file specified
at java.base/java.lang.ProcessBuilder.start(ProcessBuilder.java:1142)
at java.base/java.lang.ProcessBuilder.start(ProcessBuilder.java:1073)
at org.apache.spark.api.python.PythonWorkerFactory.createSimpleWorker(PythonWorkerFactory.scala:165)
at org.apache.spark.api.python.PythonWorkerFactory.create(PythonWorkerFactory.scala:107)
at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:119)
at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:131)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec.evaluate(BatchEvalPythonExec.scala:70)
at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$2(EvalPythonExec.scala:128)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:837)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:837)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:127)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630)
at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.io.IOException: CreateProcess error=2, The system cannot find the file specified
at java.base/java.lang.ProcessImpl.create(Native Method)
at java.base/java.lang.ProcessImpl.(ProcessImpl.java:483)
at java.base/java.lang.ProcessImpl.start(ProcessImpl.java:158)
at java.base/java.lang.ProcessBuilder.start(ProcessBuilder.java:1109)
... 26 more

Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2059)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2008)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2007)
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:2007)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:973)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:973)
at scala.Option.foreach(Option.scala:407)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:973)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2239)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2188)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2177)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:775)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2099)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2120)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2139)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:467)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:420)
at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:47)
at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3627)
at org.apache.spark.sql.Dataset.$anonfun$head$1(Dataset.scala:2697)
at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3618)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:764)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3616)
at org.apache.spark.sql.Dataset.head(Dataset.scala:2697)
at org.apache.spark.sql.Dataset.take(Dataset.scala:2904)
at org.apache.spark.sql.Dataset.getRows(Dataset.scala:300)
at org.apache.spark.sql.Dataset.showString(Dataset.scala:337)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:64)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:564)
at org.apache.spark.api.dotnet.DotnetBackendHandler.handleMethodCall(DotnetBackendHandler.scala:159)
at org.apache.spark.api.dotnet.DotnetBackendHandler.$anonfun$handleBackendRequest$2(DotnetBackendHandler.scala:99)
at org.apache.spark.api.dotnet.ThreadPool$$anon$1.run(ThreadPool.scala:34)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630)
at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.io.IOException: Cannot run program "Microsoft.Spark.Worker.exe": CreateProcess error=2, The system cannot find the file specified
at java.base/java.lang.ProcessBuilder.start(ProcessBuilder.java:1142)
at java.base/java.lang.ProcessBuilder.start(ProcessBuilder.java:1073)
at org.apache.spark.api.python.PythonWorkerFactory.createSimpleWorker(PythonWorkerFactory.scala:165)
at org.apache.spark.api.python.PythonWorkerFactory.create(PythonWorkerFactory.scala:107)
at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:119)
at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:131)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec.evaluate(BatchEvalPythonExec.scala:70)
at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$2(EvalPythonExec.scala:128)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:837)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:837)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:127)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)
... 3 more
Caused by: java.io.IOException: CreateProcess error=2, The system cannot find the file specified
at java.base/java.lang.ProcessImpl.create(Native Method)
at java.base/java.lang.ProcessImpl.(ProcessImpl.java:483)
at java.base/java.lang.ProcessImpl.start(ProcessImpl.java:158)
at java.base/java.lang.ProcessBuilder.start(ProcessBuilder.java:1109)
... 26 more

sandipbramhankar picture sandipbramhankar on 8 Feb 2021

java.io.IOException: Cannot run program "Microsoft.Spark.Worker.exe": CreateProcess error=2, The system cannot find the file specified

Can you check this doc: https://github.com/dotnet/spark/blob/master/docs/deploy-worker-udf-binaries.md?

imback82 picture imback82 on 8 Feb 2021
Was this page helpful?
0 / 5 - 0 ratings

Related issues

AFFogarty picture AFFogarty  路  6Comments
thrajput picture thrajput  路  4Comments
jonathantonic picture jonathantonic  路  9Comments
tylorhl picture tylorhl  路  4Comments
viskumar-microsoft picture viskumar-microsoft  路  7Comments