Dotnet Spark System.NullReferenceException 与 UDF

Question

我有一个简单的 Microsoft (Dotnet) Spark 应用程序，它在使用 UDF 时抛出 System.NullReferenceException。我刚刚开始使用 dotnet 学习 Spark。该应用程序看起来非常简单，但我不确定这里缺少什么

这是主要方法：

        static void Main(string[] args)
        {
            SparkSession spark = SparkSession
                .Builder()
                .AppName("Sample app")
                .GetOrCreate();
            var inputSchema = new StructType(new[]
            {
                    new StructField("f1", new StringType()),
                    new StructField("f2", new StringType())
                });
            DataFrame dataFrame = spark.Read()
                .Format("csv")
                .Option("delimiter", "\t")
                .Schema(inputSchema)
                .Load(args[0]);
            dataFrame.PrintSchema();
            var udf = Functions.Udf<string, string>(url =>
            {
                Console.WriteLine("Here");
                return "aaaa";
            });
            var df2 = dataFrame.WithColumn("f3", udf(dataFrame["f1"]));
            dataFrame.Show();
            df2.Show();
            spark.Stop();
        }

异常堆栈跟踪：

[2022-03-19T05:36:24.7336277Z] [DESKTOP-I9BK15T] [Error] [TaskRunner] [0] ProcessStream() failed with exception: System.NullReferenceException: Object reference not set to an instance of an object.
   at Microsoft.Spark.Utils.UdfSerDe.<>c.<DeserializeType>b__10_0(TypeData td) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 265
   at System.Collections.Concurrent.ConcurrentDictionary`2.GetOrAdd(TKey key, Func`2 valueFactory)
   at Microsoft.Spark.Utils.UdfSerDe.DeserializeType(TypeData typeData) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 261
   at Microsoft.Spark.Utils.UdfSerDe.Deserialize(UdfData udfData) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 160
   at Microsoft.Spark.Utils.CommandSerDe.DeserializeUdfs[T](UdfWrapperData data, Int32& nodeIndex, Int32& udfIndex) in /_/src/csharp/Microsoft.Spark/Utils/CommandSerDe.cs:line 333
   at Microsoft.Spark.Utils.CommandSerDe.Deserialize[T](Stream stream, SerializedMode& serializerMode, SerializedMode& deserializerMode, String& runMode) in /_/src/csharp/Microsoft.Spark/Utils/CommandSerDe.cs:line 306
   at Microsoft.Spark.Worker.Processor.CommandProcessor.ReadSqlCommands(PythonEvalType evalType, Stream stream) in D:\a\1\s\src\csharp\Microsoft.Spark.Worker\Processor\CommandProcessor.cs:line 188
.
.
.
   --- End of inner exception stack trace ---
   at Microsoft.Spark.Interop.Ipc.JvmBridge.CallJavaMethod(Boolean isStatic, Object classNameOrJvmObjectReference, String methodName, Object[] args)
   at Microsoft.Spark.Interop.Ipc.JvmBridge.CallNonStaticJavaMethod(JvmObjectReference jvmObject, 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 SparkJobs.Program.Main(String[] args) in C:\path\to\src\NetSpark\SparkJobs\Program.cs:line 40

输入文件 DF3.tsv

"as"    "asaaaa"    
"as"    "as"    

Spark Submit 命令

spark-submit --class org.apache.spark.deploy.dotnet.DotnetRunner --master local C:\path\to\src\NetSpark\SparkJobs\bin\Debug\netcoreapp3.1\microsoft-spark-3-0_2.12-2.0.0.jar dotnet C:\path\to\src\NetSpark\SparkJobs\bin\Debug\netcoreapp3.1\SparkJobs.dll C:\path\to\Downloads\DF3.tsv

Answer 1

我缺少以下环境变量。奇怪的是文档在安装过程中没有指定这一点。这些路径指示工作人员在哪里查找 UDF DLL 文件

set DOTNET_WORKER_DIR=C:\bin\Microsoft.Spark.Worker-1.0.0 
set DOTNET_ASSEMBLY_SEARCH_PATHS=C:\path\to\src\NetSpark\SparkJobs\bin\Debug\netcoreapp3.1

相关资源：https://learn.microsoft.com/en-us/dotnet/spark/how-to-guides/deploy-worker-udf-binaries