From c599beea8e1b6e36ac9acaee4533873aa9e591e1 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Sun, 21 Jun 2020 07:14:18 +0000
Subject: [PATCH 01/13] [fix] Fix broken maven link
---
docs/building/ubuntu-instructions.md | 18 +++++++++---------
1 file changed, 9 insertions(+), 9 deletions(-)
diff --git a/docs/building/ubuntu-instructions.md b/docs/building/ubuntu-instructions.md
index 8bb11b163..0e3dbdf40 100644
--- a/docs/building/ubuntu-instructions.md
+++ b/docs/building/ubuntu-instructions.md
@@ -35,14 +35,14 @@ If you already have all the pre-requisites, skip to the [build](ubuntu-instructi
```bash
sudo update-alternatives --config java
```
- 3. Install **[Apache Maven 3.6.0+](https://maven.apache.org/download.cgi)**
+ 3. Install **[Apache Maven 3.6.3+](https://maven.apache.org/download.cgi)**
- Run the following command:
```bash
mkdir -p ~/bin/maven
cd ~/bin/maven
- wget https://www-us.apache.org/dist/maven/maven-3/3.6.0/binaries/apache-maven-3.6.0-bin.tar.gz
- tar -xvzf apache-maven-3.6.0-bin.tar.gz
- ln -s apache-maven-3.6.0 current
+ wget https://www-us.apache.org/dist/maven/maven-3/3.6.3/binaries/apache-maven-3.6.3-bin.tar.gz
+ tar -xvzf apache-maven-3.6.3-bin.tar.gz
+ ln -s apache-maven-3.6.3 current
export M2_HOME=~/bin/maven/current
export PATH=${M2_HOME}/bin:${PATH}
source ~/.bashrc
@@ -54,11 +54,11 @@ If you already have all the pre-requisites, skip to the [build](ubuntu-instructi
<summary>📙 Click to see sample mvn -version output</summary>
```
- Apache Maven 3.6.0 (97c98ec64a1fdfee7767ce5ffb20918da4f719f3; 2018-10-24T18:41:47Z)
- Maven home: ~/bin/apache-maven-3.6.0
- Java version: 1.8.0_191, vendor: Oracle Corporation, runtime: /usr/lib/jvm/java-8-openjdk-amd64/jre
- Default locale: en, platform encoding: UTF-8
- OS name: "linux", version: "4.4.0-17763-microsoft", arch: "amd64", family: "unix"
+ Apache Maven 3.6.3 (cecedd343002696d0abb50b32b541b8a6ba2883f)
+ Maven home: ~/bin/apache-maven-3.6.3
+ Java version: 1.8.0_242, vendor: Oracle Corporation, runtime: /usr/lib/jvm/java-8-openjdk-amd64/jre
+ Default locale: en_US, platform encoding: ANSI_X3.4-1968
+ OS name: "linux", version: "4.4.0-142-generic", arch: "amd64", family: "unix"
```
4. Install **[Apache Spark 2.3+](https://spark.apache.org/downloads.html)**
- Download [Apache Spark 2.3+](https://spark.apache.org/downloads.html) and extract it into a local folder (e.g., `~/bin/spark-2.3.2-bin-hadoop2.7`)
From 394b11f0950ecf8a0b64678f71937aba80b58366 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Sun, 21 Jun 2020 14:14:41 +0000
Subject: [PATCH 02/13] Launch jvm bridge if needed at SparkSession starting
---
.../Sql/JVMBridgeHelperTests.cs | 101 +++++++++++
.../Microsoft.Spark/Sql/JVMBridgeHelper.cs | 167 ++++++++++++++++++
.../Microsoft.Spark/Sql/SparkSession.cs | 30 +++-
3 files changed, 296 insertions(+), 2 deletions(-)
create mode 100644 src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
create mode 100644 src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
diff --git a/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs b/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
new file mode 100644
index 000000000..d6bd66b33
--- /dev/null
+++ b/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
@@ -0,0 +1,101 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Net.NetworkInformation;
+using System.Net;
+using Microsoft.Spark.Interop.Ipc;
+using Microsoft.Spark.Network;
+using Microsoft.Spark.Sql;
+using Microsoft.Spark.Sql.Types;
+using Microsoft.Spark.UnitTest.TestUtils;
+using Microsoft.Spark.Utils;
+using Moq;
+using Xunit;
+
+namespace Microsoft.Spark.UnitTest
+{
+ public class JVMBridgeHelperTests
+ {
+ [Theory]
+ [InlineData("1,2,3", 0, false)]
+ [InlineData("5567,5568,5569", 0, true)]
+ [InlineData("1,2,3", 5567, false)]
+ [InlineData("5567,5568,5569", 5567, true)]
+ [InlineData("1234,5678", 1234, true)]
+ public void IsDotnetBackendPortUsingTest(string testRunningPorts, int backendport, bool expect)
+ {
+ var envkey = "DOTNETBACKEND_PORT";
+ var oldenvValue = Environment.GetEnvironmentVariable(envkey);
+ try
+ {
+ if (backendport == 0)
+ {
+ Environment.SetEnvironmentVariable(envkey, null);
+ }
+ else
+ {
+ Environment.SetEnvironmentVariable(envkey, backendport.ToString());
+ }
+
+ var runningports = testRunningPorts
+ .Split(",").Select(x => Convert.ToInt32(x)).ToList();
+ var mockIpInfos = runningports
+ .Select(p =>
+ {
+ var m_info = new Mock<TcpConnectionInformation>();
+ m_info.SetupGet(m => m.LocalEndPoint).Returns(new IPEndPoint(0, p));
+ return m_info.Object;
+ }).ToArray();
+ var ipinfo = new Mock<IPGlobalProperties>();
+ ipinfo.Setup(m => m.GetActiveTcpConnections())
+ .Returns(mockIpInfos);
+
+ var ret = JVMBridgeHelper.IsDotnetBackendPortUsing(ipinfo.Object);
+ Assert.Equal(ret, expect);
+ }
+ finally
+ {
+ Environment.SetEnvironmentVariable(envkey, oldenvValue);
+ }
+ }
+
+ /// <summary>
+ /// The main test case of JVM bridge helper,
+ /// is simply run it and close it.
+ /// </summary>
+ [Fact]
+ public void JVMBridgeHelperMainPathTest()
+ {
+ using(var helper = new JVMBridgeHelper()) {
+ // now we should be able to connect to JVM bridge
+ // or if system environment is not good, we should not failed.
+ }
+ }
+
+ /// <summary>
+ /// Test with case that already have jvm Bridge case
+ /// </summary>
+ [Fact]
+ public void JVMBridgeHelperTestsWithSparkSessionWithBridgeReady()
+ {
+ using(var helper = new JVMBridgeHelper()) {
+ // now we should be able to connect to JVM bridge
+ var spark = SparkSession.Builder().Master("local").GetOrCreate();
+ spark.Stop();
+ }
+ }
+
+ [Fact]
+ public void JVMBridgeHelperTestsWithSparkSessionWithBridgeNotReady()
+ {
+ // now we should be able to connect to JVM bridge anytime
+ var spark = SparkSession.Builder().Master("local").GetOrCreate();
+ spark.Stop();
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs b/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
new file mode 100644
index 000000000..8bf23f1ef
--- /dev/null
+++ b/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
@@ -0,0 +1,167 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.IO;
+using System.Linq;
+using System.Net.NetworkInformation;
+using System.Diagnostics;
+using System.Threading.Tasks;
+using System.Runtime.InteropServices;
+using Microsoft.Spark.Services;
+using Microsoft.Spark.Interop;
+
+namespace Microsoft.Spark.Sql
+{
+ /// <summary>
+ /// An helper to launch dotnet jvm if needed
+ /// </summary>
+ public class JVMBridgeHelper : IDisposable
+ {
+ /// <summary>
+ /// Customization for JVM Bridge jar file
+ /// If not exists, the helper will find out the jar in $DOTNET_WORKER_DIR folder.
+ /// </summary>
+ public static string JVMBridgeJarEnvName = "DOTNET_BRIDGE_JAR";
+
+ /// <summary>
+ /// DotnetRunner classname
+ /// </summary>
+ private static string RunnerClassname =
+ "org.apache.spark.deploy.dotnet.DotnetRunner";
+
+ private static string RunnerReadyMsg =
+ ".NET Backend running debug mode. Press enter to exit";
+
+ private static string RunnerAddressInUseMsg =
+ "java.net.BindException: Address already in use";
+
+
+ private static int maxWaitTimeoutMS = 60000;
+
+ /// <summary>
+ /// The running jvm bridge process , null means no such process
+ /// </summary>
+ private Process jvmBridge;
+
+ /// <summary>
+ /// Detect if we already have the runner by checking backend port is using or not.
+ /// </summary>
+ /// <param name="customIPGlobalProperties">custom IPGlobalProperties, null for System.Net.NetworkInformation</param>
+ /// <returns> True means backend port is occupied by the runner.</returns>
+ public static bool IsDotnetBackendPortUsing(
+ IPGlobalProperties customIPGlobalProperties = null)
+ {
+ var backendport = SparkEnvironment.ConfigurationService.GetBackendPortNumber();
+ var activeTcps =
+ (customIPGlobalProperties ?? IPGlobalProperties.GetIPGlobalProperties())
+ .GetActiveTcpConnections();
+ return activeTcps.Any(tcp => tcp.LocalEndPoint.Port == backendport);
+ }
+
+ public JVMBridgeHelper()
+ {
+ var jarpath = locateBridgeJar();
+ var sparksubmit = locateSparkSubmit();
+ if (string.IsNullOrWhiteSpace(jarpath) ||
+ string.IsNullOrWhiteSpace(sparksubmit))
+ {
+ // Cannot find correct launch informations, give up.
+ return;
+ }
+ var arguments = $"--class {RunnerClassname} {jarpath} debug";
+ var startupinfo = new ProcessStartInfo
+ {
+ FileName = sparksubmit,
+ Arguments = arguments,
+ RedirectStandardOutput = true,
+ RedirectStandardInput = true,
+ UseShellExecute = false,
+ CreateNoWindow = true,
+ };
+
+ jvmBridge = new Process() { StartInfo = startupinfo };
+ jvmBridge.Start();
+
+ // wait until we see .net backend started
+ Task<string> message;
+ while ((message = jvmBridge.StandardOutput.ReadLineAsync()) != null)
+ {
+ if (message.Wait(maxWaitTimeoutMS) == false)
+ {
+ // wait timeout , giveup
+ break;
+ }
+
+ if (message.Result.Contains(RunnerReadyMsg))
+ {
+ // launched successfully!
+ return;
+ }
+ if (message.Result.Contains(RunnerAddressInUseMsg))
+ {
+ // failed to start for port is using, give up.
+ break;
+ }
+ }
+ // wait timeout , or failed to startup
+ // give up.
+ jvmBridge.Close();
+ jvmBridge = null;
+ }
+
+ private string locateSparkSubmit()
+ {
+ var sparkHome = Environment.GetEnvironmentVariable("SPARK_HOME");
+ var filename = Path.Combine(sparkHome, "bin", "spark-submit");
+ if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+ {
+ filename += ".cmd";
+ }
+
+ if (!File.Exists(filename))
+ {
+ return string.Empty;
+ }
+
+ return filename;
+ }
+
+ private string locateBridgeJar()
+ {
+ var jarpath = Environment.GetEnvironmentVariable(JVMBridgeJarEnvName);
+ if (string.IsNullOrWhiteSpace(jarpath) == false)
+ {
+ return jarpath;
+ }
+
+ var workdir = Environment.GetEnvironmentVariable(
+ ConfigurationService.WorkerDirEnvVarName);
+ if ((workdir != null) && Directory.Exists(workdir))
+ {
+ // let's find the approicate jar in the work dirctory.
+ var jarfile = new DirectoryInfo(workdir)
+ .GetFiles("microsoft-spark-*.jar")
+ .FirstOrDefault();
+ if (jarfile != null)
+ {
+ return Path.Combine(jarfile.DirectoryName, jarfile.Name);
+ }
+ }
+
+ return string.Empty;
+ }
+
+ public void Dispose()
+ {
+ if (jvmBridge != null)
+ {
+ jvmBridge.StandardInput.WriteLine("\n");
+ // to avoid deadlock, read all output then wait for exit.
+ jvmBridge.StandardOutput.ReadToEndAsync();
+ jvmBridge.WaitForExit(maxWaitTimeoutMS);
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/csharp/Microsoft.Spark/Sql/SparkSession.cs b/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
index fc706081f..c9be02151 100644
--- a/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
+++ b/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
@@ -27,6 +27,12 @@ public sealed class SparkSession : IDisposable, IJvmObjectReferenceProvider
private static readonly string s_sparkSessionClassName =
"org.apache.spark.sql.SparkSession";
+ /// <summary>
+ /// The created jvm process
+ /// </summary>
+ /// <returns>The created jvm bridge process helper.</returns>
+ private static JVMBridgeHelper s_jvmbridge = null;
+
/// <summary>
/// Constructor for SparkSession.
/// </summary>
@@ -59,7 +65,17 @@ internal SparkSession(JvmObjectReference jvmObject)
/// Creates a Builder object for SparkSession.
/// </summary>
/// <returns>Builder object</returns>
- public static Builder Builder() => new Builder();
+ public static Builder Builder()
+ {
+ // We could try to detect if we don't have jvm bridge,
+ // call the helper to launch jvm bridge process.
+ if ((s_jvmbridge == null) &&
+ (JVMBridgeHelper.IsDotnetBackendPortUsing() == false))
+ {
+ s_jvmbridge = new JVMBridgeHelper();
+ }
+ return new Builder();
+ }
/// Note that *ActiveSession() APIs are not exposed because these APIs work with a
/// thread-local variable, which stores the session variable. Since the Netty server
@@ -293,7 +309,17 @@ public UdfRegistration Udf() =>
/// <summary>
/// Stops the underlying SparkContext.
/// </summary>
- public void Stop() => _jvmObject.Invoke("stop");
+ public void Stop()
+ {
+ _jvmObject.Invoke("stop");
+
+ // if we have created the jvm bridge process, dispose it now.
+ if (s_jvmbridge != null)
+ {
+ s_jvmbridge.Dispose();
+ s_jvmbridge = null;
+ }
+ }
/// <summary>
/// Returns a single column schema of the given datatype.
From c376c0e5e83444af52d6c4725bdc9471b2004635 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Sun, 21 Jun 2020 16:54:42 +0000
Subject: [PATCH 03/13] [fix] Add more test and fix Listening TCP issues.
---
.../Sql/JVMBridgeHelperTests.cs | 30 +++++++++++--------
.../Microsoft.Spark/Sql/JVMBridgeHelper.cs | 16 +++++-----
2 files changed, 26 insertions(+), 20 deletions(-)
diff --git a/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs b/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
index d6bd66b33..daa05133d 100644
--- a/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
+++ b/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
@@ -27,7 +27,7 @@ public class JVMBridgeHelperTests
[InlineData("1,2,3", 5567, false)]
[InlineData("5567,5568,5569", 5567, true)]
[InlineData("1234,5678", 1234, true)]
- public void IsDotnetBackendPortUsingTest(string testRunningPorts, int backendport, bool expect)
+ public void IsDotnetBackendPortUsingTest(string testListeningPorts, int backendport, bool expect)
{
var envkey = "DOTNETBACKEND_PORT";
var oldenvValue = Environment.GetEnvironmentVariable(envkey);
@@ -42,18 +42,12 @@ public void IsDotnetBackendPortUsingTest(string testRunningPorts, int backendpor
Environment.SetEnvironmentVariable(envkey, backendport.ToString());
}
- var runningports = testRunningPorts
- .Split(",").Select(x => Convert.ToInt32(x)).ToList();
- var mockIpInfos = runningports
- .Select(p =>
- {
- var m_info = new Mock<TcpConnectionInformation>();
- m_info.SetupGet(m => m.LocalEndPoint).Returns(new IPEndPoint(0, p));
- return m_info.Object;
- }).ToArray();
+ var listeningEndpoints = testListeningPorts
+ .Split(",").Select(x => new IPEndPoint(0, Convert.ToInt32(x))).ToArray();
+
var ipinfo = new Mock<IPGlobalProperties>();
- ipinfo.Setup(m => m.GetActiveTcpConnections())
- .Returns(mockIpInfos);
+ ipinfo.Setup(m => m.GetActiveTcpListeners())
+ .Returns(listeningEndpoints);
var ret = JVMBridgeHelper.IsDotnetBackendPortUsing(ipinfo.Object);
Assert.Equal(ret, expect);
@@ -75,7 +69,17 @@ public void JVMBridgeHelperMainPathTest()
// now we should be able to connect to JVM bridge
// or if system environment is not good, we should not failed.
}
- }
+ }
+
+ [Fact]
+ public void JVMBridgeHelperWithoutSpark()
+ {
+ var oldhome = Environment.GetEnvironmentVariable("SPARK_HOME");
+ Environment.SetEnvironmentVariable("SPARK_HOME", null);
+ using(var helper = new JVMBridgeHelper()) {
+ }
+ Environment.SetEnvironmentVariable("SPARK_HOME", oldhome);
+ }
/// <summary>
/// Test with case that already have jvm Bridge case
diff --git a/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs b/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
index 8bf23f1ef..ce5de0888 100644
--- a/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
+++ b/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
@@ -54,10 +54,10 @@ public static bool IsDotnetBackendPortUsing(
IPGlobalProperties customIPGlobalProperties = null)
{
var backendport = SparkEnvironment.ConfigurationService.GetBackendPortNumber();
- var activeTcps =
+ var listeningEndpoints =
(customIPGlobalProperties ?? IPGlobalProperties.GetIPGlobalProperties())
- .GetActiveTcpConnections();
- return activeTcps.Any(tcp => tcp.LocalEndPoint.Port == backendport);
+ .GetActiveTcpListeners();
+ return listeningEndpoints.Any(p => p.Port == backendport);
}
public JVMBridgeHelper()
@@ -97,11 +97,13 @@ public JVMBridgeHelper()
if (message.Result.Contains(RunnerReadyMsg))
{
// launched successfully!
+ jvmBridge.StandardOutput.ReadToEndAsync();
return;
}
if (message.Result.Contains(RunnerAddressInUseMsg))
{
// failed to start for port is using, give up.
+ jvmBridge.StandardOutput.ReadToEndAsync();
break;
}
}
@@ -114,17 +116,19 @@ public JVMBridgeHelper()
private string locateSparkSubmit()
{
var sparkHome = Environment.GetEnvironmentVariable("SPARK_HOME");
+ if (string.IsNullOrWhiteSpace(sparkHome))
+ {
+ return string.Empty;
+ }
var filename = Path.Combine(sparkHome, "bin", "spark-submit");
if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
{
filename += ".cmd";
}
-
if (!File.Exists(filename))
{
return string.Empty;
}
-
return filename;
}
@@ -158,8 +162,6 @@ public void Dispose()
if (jvmBridge != null)
{
jvmBridge.StandardInput.WriteLine("\n");
- // to avoid deadlock, read all output then wait for exit.
- jvmBridge.StandardOutput.ReadToEndAsync();
jvmBridge.WaitForExit(maxWaitTimeoutMS);
}
}
From a07e7817cec4a1a4c40518abcc4071f12e7ce18f Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Mon, 22 Jun 2020 01:32:10 +0000
Subject: [PATCH 04/13] Fix broken maven link in window build doc
---
docs/building/windows-instructions.md | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/docs/building/windows-instructions.md b/docs/building/windows-instructions.md
index 84874a129..aad141b68 100644
--- a/docs/building/windows-instructions.md
+++ b/docs/building/windows-instructions.md
@@ -30,10 +30,10 @@ If you already have all the pre-requisites, skip to the [build](windows-instruct
3. Install **[Java 1.8](https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html)**
- Select the appropriate version for your operating system e.g., jdk-8u201-windows-x64.exe for Win x64 machine.
- Install using the installer and verify you are able to run `java` from your command-line
- 4. Install **[Apache Maven 3.6.0+](https://maven.apache.org/download.cgi)**
- - Download [Apache Maven 3.6.0](http://mirror.metrocast.net/apache/maven/maven-3/3.6.0/binaries/apache-maven-3.6.0-bin.zip)
- - Extract to a local directory e.g., `c:\bin\apache-maven-3.6.0\`
- - Add Apache Maven to your [PATH environment variable](https://www.java.com/en/download/help/path.xml) e.g., `c:\bin\apache-maven-3.6.0\bin`
+ 4. Install **[Apache Maven 3.6.3+](https://maven.apache.org/download.cgi)**
+ - Download [Apache Maven 3.6.3](http://mirror.metrocast.net/apache/maven/maven-3/3.6.3/binaries/apache-maven-3.6.3-bin.zip)
+ - Extract to a local directory e.g., `c:\bin\apache-maven-3.6.3\`
+ - Add Apache Maven to your [PATH environment variable](https://www.java.com/en/download/help/path.xml) e.g., `c:\bin\apache-maven-3.6.3\bin`
- Verify you are able to run `mvn` from your command-line
5. Install **[Apache Spark 2.3+](https://spark.apache.org/downloads.html)**
- Download [Apache Spark 2.3+](https://spark.apache.org/downloads.html) and extract it into a local folder (e.g., `c:\bin\spark-2.3.2-bin-hadoop2.7\`) using [7-zip](https://www.7-zip.org/).
From ffabc5335443b6eb2e1de88c18abeb5046eb56c0 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Mon, 22 Jun 2020 10:00:52 +0000
Subject: [PATCH 05/13] Remove integration test for unit test may not pass at
env not ready.
---
.../Sql/JVMBridgeHelperTests.cs | 29 ++++---------------
1 file changed, 5 insertions(+), 24 deletions(-)
diff --git a/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs b/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
index daa05133d..a2b6ad2ca 100644
--- a/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
+++ b/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
@@ -63,9 +63,10 @@ public void IsDotnetBackendPortUsingTest(string testListeningPorts, int backendp
/// is simply run it and close it.
/// </summary>
[Fact]
- public void JVMBridgeHelperMainPathTest()
+ public void JVMBridgeHelperMainPathTest()
{
- using(var helper = new JVMBridgeHelper()) {
+ using (var helper = new JVMBridgeHelper())
+ {
// now we should be able to connect to JVM bridge
// or if system environment is not good, we should not failed.
}
@@ -76,30 +77,10 @@ public void JVMBridgeHelperWithoutSpark()
{
var oldhome = Environment.GetEnvironmentVariable("SPARK_HOME");
Environment.SetEnvironmentVariable("SPARK_HOME", null);
- using(var helper = new JVMBridgeHelper()) {
+ using (var helper = new JVMBridgeHelper())
+ {
}
Environment.SetEnvironmentVariable("SPARK_HOME", oldhome);
}
-
- /// <summary>
- /// Test with case that already have jvm Bridge case
- /// </summary>
- [Fact]
- public void JVMBridgeHelperTestsWithSparkSessionWithBridgeReady()
- {
- using(var helper = new JVMBridgeHelper()) {
- // now we should be able to connect to JVM bridge
- var spark = SparkSession.Builder().Master("local").GetOrCreate();
- spark.Stop();
- }
- }
-
- [Fact]
- public void JVMBridgeHelperTestsWithSparkSessionWithBridgeNotReady()
- {
- // now we should be able to connect to JVM bridge anytime
- var spark = SparkSession.Builder().Master("local").GetOrCreate();
- spark.Stop();
- }
}
}
\ No newline at end of file
From 32d50aafa229c7c8dabbd71517a8b7d5d2ce1944 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Mon, 22 Jun 2020 14:56:15 +0000
Subject: [PATCH 06/13] Add logger to show jvm bridge launch info.
---
src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs | 6 ++++++
1 file changed, 6 insertions(+)
diff --git a/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs b/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
index ce5de0888..21c065654 100644
--- a/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
+++ b/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
@@ -40,6 +40,9 @@ public class JVMBridgeHelper : IDisposable
private static int maxWaitTimeoutMS = 60000;
+ private readonly ILoggerService _logger =
+ LoggerServiceFactory.GetLogger(typeof(JVMBridgeHelper));
+
/// <summary>
/// The running jvm bridge process , null means no such process
/// </summary>
@@ -82,6 +85,7 @@ public JVMBridgeHelper()
};
jvmBridge = new Process() { StartInfo = startupinfo };
+ _logger.LogInfo($"Launch JVM Bridge : {sparksubmit} {arguments}");
jvmBridge.Start();
// wait until we see .net backend started
@@ -98,6 +102,7 @@ public JVMBridgeHelper()
{
// launched successfully!
jvmBridge.StandardOutput.ReadToEndAsync();
+ _logger.LogInfo($"Launch JVM Bridge ready");
return;
}
if (message.Result.Contains(RunnerAddressInUseMsg))
@@ -163,6 +168,7 @@ public void Dispose()
{
jvmBridge.StandardInput.WriteLine("\n");
jvmBridge.WaitForExit(maxWaitTimeoutMS);
+ _logger.LogInfo($"JVM Bridge disposed.");
}
}
}
From 70add1d1771d447b703a615cad4c2e0ac4677579 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Mon, 22 Jun 2020 14:56:15 +0000
Subject: [PATCH 07/13] Add logger to show jvm bridge launch info.
---
.vscode/launch.json | 27 ++++++++++++
.vscode/settings.json | 3 ++
.vscode/tasks.json | 42 +++++++++++++++++++
.../Microsoft.Spark/Sql/Types/ComplexTypes.cs | 33 +++++++++++++--
4 files changed, 101 insertions(+), 4 deletions(-)
create mode 100644 .vscode/launch.json
create mode 100644 .vscode/settings.json
create mode 100644 .vscode/tasks.json
diff --git a/.vscode/launch.json b/.vscode/launch.json
new file mode 100644
index 000000000..4e219e522
--- /dev/null
+++ b/.vscode/launch.json
@@ -0,0 +1,27 @@
+{
+ // Use IntelliSense to find out which attributes exist for C# debugging
+ // Use hover for the description of the existing attributes
+ // For further information visit https://github.com/OmniSharp/omnisharp-vscode/blob/master/debugger-launchjson.md
+ "version": "0.2.0",
+ "configurations": [
+ {
+ "name": ".NET Core Launch (console)",
+ "type": "coreclr",
+ "request": "launch",
+ "preLaunchTask": "build",
+ // If you have changed target frameworks, make sure to update the program path.
+ "program": "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker/bin/Debug/netcoreapp3.1/Microsoft.Spark.Worker.dll",
+ "args": [],
+ "cwd": "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker",
+ // For more information about the 'console' field, see https://aka.ms/VSCode-CS-LaunchJson-Console
+ "console": "internalConsole",
+ "stopAtEntry": false
+ },
+ {
+ "name": ".NET Core Attach",
+ "type": "coreclr",
+ "request": "attach",
+ "processId": "${command:pickProcess}"
+ }
+ ]
+}
\ No newline at end of file
diff --git a/.vscode/settings.json b/.vscode/settings.json
new file mode 100644
index 000000000..d6ec7df72
--- /dev/null
+++ b/.vscode/settings.json
@@ -0,0 +1,3 @@
+{
+ "dotnet-test-explorer.testProjectPath": "src/csharp/Microsoft.Spark.UnitTest"
+}
\ No newline at end of file
diff --git a/.vscode/tasks.json b/.vscode/tasks.json
new file mode 100644
index 000000000..0f8011d05
--- /dev/null
+++ b/.vscode/tasks.json
@@ -0,0 +1,42 @@
+{
+ "version": "2.0.0",
+ "tasks": [
+ {
+ "label": "build",
+ "command": "dotnet",
+ "type": "process",
+ "args": [
+ "build",
+ "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker/Microsoft.Spark.Worker.csproj",
+ "/property:GenerateFullPaths=true",
+ "/consoleloggerparameters:NoSummary"
+ ],
+ "problemMatcher": "$msCompile"
+ },
+ {
+ "label": "publish",
+ "command": "dotnet",
+ "type": "process",
+ "args": [
+ "publish",
+ "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker/Microsoft.Spark.Worker.csproj",
+ "/property:GenerateFullPaths=true",
+ "/consoleloggerparameters:NoSummary"
+ ],
+ "problemMatcher": "$msCompile"
+ },
+ {
+ "label": "watch",
+ "command": "dotnet",
+ "type": "process",
+ "args": [
+ "watch",
+ "run",
+ "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker/Microsoft.Spark.Worker.csproj",
+ "/property:GenerateFullPaths=true",
+ "/consoleloggerparameters:NoSummary"
+ ],
+ "problemMatcher": "$msCompile"
+ }
+ ]
+}
\ No newline at end of file
diff --git a/src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs b/src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs
index 2b65ea6d1..2fa5f4f2f 100644
--- a/src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs
+++ b/src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs
@@ -69,9 +69,22 @@ private DataType FromJson(JObject json)
return this;
}
- internal override bool NeedConversion() => true;
+ internal override bool NeedConversion() => ElementType.NeedConversion();
- internal override object FromInternal(object obj) => throw new NotImplementedException();
+ internal override object FromInternal(object obj)
+ {
+ if (!NeedConversion())
+ {
+ return obj;
+ }
+
+ var convertedObj = new List<object>();
+ foreach(object o in (dynamic)obj)
+ {
+ convertedObj.Add(ElementType.FromInternal(o));
+ }
+ return convertedObj;
+ }
}
/// <summary>
@@ -142,9 +155,21 @@ private DataType FromJson(JObject json)
return this;
}
- internal override bool NeedConversion() => true;
+ internal override bool NeedConversion() => KeyType.NeedConversion() && ValueType.NeedConversion();
- internal override object FromInternal(object obj) => throw new NotImplementedException();
+ internal override object FromInternal(object obj)
+ {
+ if (!NeedConversion())
+ {
+ return obj;
+ }
+ var convertedObj = new Dictionary<object, object>();
+ foreach(dynamic kv in (dynamic)obj)
+ {
+ convertedObj[kv.Key] = kv.Value;
+ }
+ return convertedObj;
+ }
}
/// <summary>
From 790bb093384a4debbce994c72e515db83e3eeb8a Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Mon, 22 Jun 2020 15:01:13 +0000
Subject: [PATCH 08/13] Revert "Add logger to show jvm bridge launch info."
This reverts commit 70add1d1771d447b703a615cad4c2e0ac4677579.
---
.vscode/launch.json | 27 ------------
.vscode/settings.json | 3 --
.vscode/tasks.json | 42 -------------------
.../Microsoft.Spark/Sql/Types/ComplexTypes.cs | 33 ++-------------
4 files changed, 4 insertions(+), 101 deletions(-)
delete mode 100644 .vscode/launch.json
delete mode 100644 .vscode/settings.json
delete mode 100644 .vscode/tasks.json
diff --git a/.vscode/launch.json b/.vscode/launch.json
deleted file mode 100644
index 4e219e522..000000000
--- a/.vscode/launch.json
+++ /dev/null
@@ -1,27 +0,0 @@
-{
- // Use IntelliSense to find out which attributes exist for C# debugging
- // Use hover for the description of the existing attributes
- // For further information visit https://github.com/OmniSharp/omnisharp-vscode/blob/master/debugger-launchjson.md
- "version": "0.2.0",
- "configurations": [
- {
- "name": ".NET Core Launch (console)",
- "type": "coreclr",
- "request": "launch",
- "preLaunchTask": "build",
- // If you have changed target frameworks, make sure to update the program path.
- "program": "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker/bin/Debug/netcoreapp3.1/Microsoft.Spark.Worker.dll",
- "args": [],
- "cwd": "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker",
- // For more information about the 'console' field, see https://aka.ms/VSCode-CS-LaunchJson-Console
- "console": "internalConsole",
- "stopAtEntry": false
- },
- {
- "name": ".NET Core Attach",
- "type": "coreclr",
- "request": "attach",
- "processId": "${command:pickProcess}"
- }
- ]
-}
\ No newline at end of file
diff --git a/.vscode/settings.json b/.vscode/settings.json
deleted file mode 100644
index d6ec7df72..000000000
--- a/.vscode/settings.json
+++ /dev/null
@@ -1,3 +0,0 @@
-{
- "dotnet-test-explorer.testProjectPath": "src/csharp/Microsoft.Spark.UnitTest"
-}
\ No newline at end of file
diff --git a/.vscode/tasks.json b/.vscode/tasks.json
deleted file mode 100644
index 0f8011d05..000000000
--- a/.vscode/tasks.json
+++ /dev/null
@@ -1,42 +0,0 @@
-{
- "version": "2.0.0",
- "tasks": [
- {
- "label": "build",
- "command": "dotnet",
- "type": "process",
- "args": [
- "build",
- "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker/Microsoft.Spark.Worker.csproj",
- "/property:GenerateFullPaths=true",
- "/consoleloggerparameters:NoSummary"
- ],
- "problemMatcher": "$msCompile"
- },
- {
- "label": "publish",
- "command": "dotnet",
- "type": "process",
- "args": [
- "publish",
- "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker/Microsoft.Spark.Worker.csproj",
- "/property:GenerateFullPaths=true",
- "/consoleloggerparameters:NoSummary"
- ],
- "problemMatcher": "$msCompile"
- },
- {
- "label": "watch",
- "command": "dotnet",
- "type": "process",
- "args": [
- "watch",
- "run",
- "${workspaceFolder}/src/csharp/Microsoft.Spark.Worker/Microsoft.Spark.Worker.csproj",
- "/property:GenerateFullPaths=true",
- "/consoleloggerparameters:NoSummary"
- ],
- "problemMatcher": "$msCompile"
- }
- ]
-}
\ No newline at end of file
diff --git a/src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs b/src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs
index 2fa5f4f2f..2b65ea6d1 100644
--- a/src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs
+++ b/src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs
@@ -69,22 +69,9 @@ private DataType FromJson(JObject json)
return this;
}
- internal override bool NeedConversion() => ElementType.NeedConversion();
+ internal override bool NeedConversion() => true;
- internal override object FromInternal(object obj)
- {
- if (!NeedConversion())
- {
- return obj;
- }
-
- var convertedObj = new List<object>();
- foreach(object o in (dynamic)obj)
- {
- convertedObj.Add(ElementType.FromInternal(o));
- }
- return convertedObj;
- }
+ internal override object FromInternal(object obj) => throw new NotImplementedException();
}
/// <summary>
@@ -155,21 +142,9 @@ private DataType FromJson(JObject json)
return this;
}
- internal override bool NeedConversion() => KeyType.NeedConversion() && ValueType.NeedConversion();
+ internal override bool NeedConversion() => true;
- internal override object FromInternal(object obj)
- {
- if (!NeedConversion())
- {
- return obj;
- }
- var convertedObj = new Dictionary<object, object>();
- foreach(dynamic kv in (dynamic)obj)
- {
- convertedObj[kv.Key] = kv.Value;
- }
- return convertedObj;
- }
+ internal override object FromInternal(object obj) => throw new NotImplementedException();
}
/// <summary>
From 28ef78a5cb61072e245346379485d67301c8de85 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Wed, 24 Jun 2020 13:02:00 +0000
Subject: [PATCH 09/13] Add logs for spark fixture processing
---
src/csharp/Microsoft.Spark.E2ETest/SparkFixture.cs | 8 ++++++++
1 file changed, 8 insertions(+)
diff --git a/src/csharp/Microsoft.Spark.E2ETest/SparkFixture.cs b/src/csharp/Microsoft.Spark.E2ETest/SparkFixture.cs
index 6d8dadbac..eda1400a1 100644
--- a/src/csharp/Microsoft.Spark.E2ETest/SparkFixture.cs
+++ b/src/csharp/Microsoft.Spark.E2ETest/SparkFixture.cs
@@ -9,6 +9,7 @@
using System.Runtime.InteropServices;
using Microsoft.Spark.Interop.Ipc;
using Microsoft.Spark.Sql;
+using Microsoft.Spark.Services;
using Microsoft.Spark.UnitTest.TestUtils;
using Xunit;
@@ -21,6 +22,10 @@ namespace Microsoft.Spark.E2ETest
/// </summary>
public sealed class SparkFixture : IDisposable
{
+
+ private readonly ILoggerService _logger =
+ LoggerServiceFactory.GetLogger(typeof(SparkFixture));
+
/// <summary>
/// The names of environment variables used by the SparkFixture.
/// </summary>
@@ -112,11 +117,14 @@ public SparkFixture()
Spark.SparkContext.SetLogLevel(DefaultLogLevel);
Jvm = ((IJvmObjectReferenceProvider)Spark).Reference.Jvm;
+
+ _logger.LogInfo("SparkSession created.");
}
public void Dispose()
{
Spark.Dispose();
+ _logger.LogInfo("SparkSession disposed.");
// CSparkRunner will exit upon receiving newline from
// the standard input stream.
From fc2442cb3acb23274c00a460aba825da35b70325 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Thu, 25 Jun 2020 05:04:57 +0000
Subject: [PATCH 10/13] Move JVMBridgeHelper to Utils and try to locate
SparkHome
---
.../{Sql => Utils}/JVMBridgeHelperTests.cs | 0
.../Microsoft.Spark/Sql/SparkSession.cs | 1 +
.../{Sql => Utils}/JVMBridgeHelper.cs | 44 ++----
.../Microsoft.Spark/Utils/SparkSettings.cs | 149 ++++++++++++++++++
4 files changed, 166 insertions(+), 28 deletions(-)
rename src/csharp/Microsoft.Spark.UnitTest/{Sql => Utils}/JVMBridgeHelperTests.cs (100%)
rename src/csharp/Microsoft.Spark/{Sql => Utils}/JVMBridgeHelper.cs (83%)
create mode 100644 src/csharp/Microsoft.Spark/Utils/SparkSettings.cs
diff --git a/src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs b/src/csharp/Microsoft.Spark.UnitTest/Utils/JVMBridgeHelperTests.cs
similarity index 100%
rename from src/csharp/Microsoft.Spark.UnitTest/Sql/JVMBridgeHelperTests.cs
rename to src/csharp/Microsoft.Spark.UnitTest/Utils/JVMBridgeHelperTests.cs
diff --git a/src/csharp/Microsoft.Spark/Sql/SparkSession.cs b/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
index d524185b3..cebc1e85b 100644
--- a/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
+++ b/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
@@ -11,6 +11,7 @@
using Microsoft.Spark.Interop.Ipc;
using Microsoft.Spark.Sql.Streaming;
using Microsoft.Spark.Sql.Types;
+using Microsoft.Spark.Utils;
namespace Microsoft.Spark.Sql
{
diff --git a/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs b/src/csharp/Microsoft.Spark/Utils/JVMBridgeHelper.cs
similarity index 83%
rename from src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
rename to src/csharp/Microsoft.Spark/Utils/JVMBridgeHelper.cs
index 21c065654..cb541115e 100644
--- a/src/csharp/Microsoft.Spark/Sql/JVMBridgeHelper.cs
+++ b/src/csharp/Microsoft.Spark/Utils/JVMBridgeHelper.cs
@@ -12,18 +12,23 @@
using Microsoft.Spark.Services;
using Microsoft.Spark.Interop;
-namespace Microsoft.Spark.Sql
+namespace Microsoft.Spark.Utils
{
/// <summary>
/// An helper to launch dotnet jvm if needed
/// </summary>
- public class JVMBridgeHelper : IDisposable
+ internal class JVMBridgeHelper : IDisposable
{
/// <summary>
/// Customization for JVM Bridge jar file
/// If not exists, the helper will find out the jar in $DOTNET_WORKER_DIR folder.
/// </summary>
- public static string JVMBridgeJarEnvName = "DOTNET_BRIDGE_JAR";
+ internal static string JVMBridgeJarEnvName = "DOTNET_BRIDGE_JAR";
+
+ /// <summary>
+ /// Generate spark settings for the running system
+ /// </summary>
+ internal static SparkSettings sparkSettings = new SparkSettings();
/// <summary>
/// DotnetRunner classname
@@ -53,7 +58,7 @@ public class JVMBridgeHelper : IDisposable
/// </summary>
/// <param name="customIPGlobalProperties">custom IPGlobalProperties, null for System.Net.NetworkInformation</param>
/// <returns> True means backend port is occupied by the runner.</returns>
- public static bool IsDotnetBackendPortUsing(
+ internal static bool IsDotnetBackendPortUsing(
IPGlobalProperties customIPGlobalProperties = null)
{
var backendport = SparkEnvironment.ConfigurationService.GetBackendPortNumber();
@@ -63,12 +68,11 @@ public static bool IsDotnetBackendPortUsing(
return listeningEndpoints.Any(p => p.Port == backendport);
}
- public JVMBridgeHelper()
+ internal JVMBridgeHelper()
{
var jarpath = locateBridgeJar();
- var sparksubmit = locateSparkSubmit();
if (string.IsNullOrWhiteSpace(jarpath) ||
- string.IsNullOrWhiteSpace(sparksubmit))
+ string.IsNullOrWhiteSpace(sparkSettings.SPARK_SUBMIT))
{
// Cannot find correct launch informations, give up.
return;
@@ -76,7 +80,7 @@ public JVMBridgeHelper()
var arguments = $"--class {RunnerClassname} {jarpath} debug";
var startupinfo = new ProcessStartInfo
{
- FileName = sparksubmit,
+ FileName = sparkSettings.SPARK_SUBMIT,
Arguments = arguments,
RedirectStandardOutput = true,
RedirectStandardInput = true,
@@ -85,7 +89,7 @@ public JVMBridgeHelper()
};
jvmBridge = new Process() { StartInfo = startupinfo };
- _logger.LogInfo($"Launch JVM Bridge : {sparksubmit} {arguments}");
+ _logger.LogInfo($"Launch JVM Bridge : {sparkSettings.SPARK_SUBMIT} {arguments}");
jvmBridge.Start();
// wait until we see .net backend started
@@ -118,25 +122,9 @@ public JVMBridgeHelper()
jvmBridge = null;
}
- private string locateSparkSubmit()
- {
- var sparkHome = Environment.GetEnvironmentVariable("SPARK_HOME");
- if (string.IsNullOrWhiteSpace(sparkHome))
- {
- return string.Empty;
- }
- var filename = Path.Combine(sparkHome, "bin", "spark-submit");
- if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
- {
- filename += ".cmd";
- }
- if (!File.Exists(filename))
- {
- return string.Empty;
- }
- return filename;
- }
-
+ /// <summary>
+ /// Locate
+ /// </summary>
private string locateBridgeJar()
{
var jarpath = Environment.GetEnvironmentVariable(JVMBridgeJarEnvName);
diff --git a/src/csharp/Microsoft.Spark/Utils/SparkSettings.cs b/src/csharp/Microsoft.Spark/Utils/SparkSettings.cs
new file mode 100644
index 000000000..09a8c1d06
--- /dev/null
+++ b/src/csharp/Microsoft.Spark/Utils/SparkSettings.cs
@@ -0,0 +1,149 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.IO;
+using System.Linq;
+using System.Collections.Generic;
+using System.Runtime.InteropServices;
+
+namespace Microsoft.Spark.Utils
+{
+ /// <summary>
+ /// Discover the spark settings
+ /// </summary>
+ internal class SparkSettings
+ {
+ private static string SPARK_HOME_ENV_KEY = "SPARK_HOME";
+
+ /// <summary>
+ /// The spark home
+ /// This would be empty string when failed to findout.
+ /// </summary>
+ public string SPARK_HOME {get; private set;}
+
+ /// <summary>
+ /// The spark submit file
+ /// </summary>
+ public string SPARK_SUBMIT {get;private set;}
+
+
+ public Version Version { get; private set; }
+
+ private static string sparksubmitcmd =
+ RuntimeInformation.IsOSPlatform(OSPlatform.Windows)
+ ? "spark-submit.cmd" : "spark-submit";
+
+ /// <summary>
+ /// Init the spark settings
+ /// </summary>
+ public SparkSettings()
+ {
+ find_spark_home();
+ this.SPARK_SUBMIT = locateSparkSubmit(SPARK_HOME);
+ InitVersion();
+ }
+
+ /// <summary>
+ /// Locate SPARK_HOME path
+ /// </summary>
+ /// <returns>Return empty string when failed to find out.</returns>
+ private void find_spark_home()
+ {
+ SPARK_HOME = Environment.GetEnvironmentVariable(SPARK_HOME_ENV_KEY);
+ if (string.IsNullOrWhiteSpace(SPARK_HOME) == false)
+ {
+ return;
+ }
+ foreach (var possiblehome in possible_spark_home())
+ {
+ if (isSparkHome(possiblehome))
+ {
+ SPARK_HOME = possiblehome;
+ Environment.SetEnvironmentVariable(SPARK_HOME_ENV_KEY, SPARK_HOME);
+ return;
+ }
+ }
+ }
+
+ private IEnumerable<string> possible_spark_home()
+ {
+ var pathSeperator = ':';
+ var envPath = "PATH";
+ if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+ {
+ pathSeperator = ';';
+ envPath = "Path";
+ }
+ // try Env Pathes if we could locate spark
+ foreach (var path in Environment.GetEnvironmentVariable(envPath).Split(pathSeperator))
+ {
+ yield return path;
+ yield return Path.GetFullPath(Path.Combine(path, ".."));
+ }
+ }
+
+ /// <summary>
+ /// find out the existing spark-submit file
+ /// </summary>
+ private string locateSparkSubmit(string sparkhome)
+ {
+ var fn = Path.Combine(sparkhome, "bin", sparksubmitcmd);
+ return File.Exists(fn)
+ ? fn : string.Empty;
+ }
+
+ private string locateJarFolder(string sparkhome)
+ {
+ var possible = new string[] { "jars", "assembly" };
+ foreach (var tryfolder in possible)
+ {
+ var folder = Path.Combine(sparkhome, tryfolder);
+ if (Directory.Exists(folder))
+ {
+ return folder;
+ }
+ }
+ return string.Empty;
+ }
+
+ /// <summary>
+ /// Check if it is a reasonable SPARK_HOME
+ /// </summary>
+ private bool isSparkHome(string path)
+ {
+ return (locateSparkSubmit(path) != string.Empty) &&
+ (locateJarFolder(path) != string.Empty);
+ }
+
+ /// <summary>
+ /// After init spark home, try to find out spark version
+ /// </summary>
+ private void InitVersion()
+ {
+ var releasefile = $"{SPARK_HOME}{Path.DirectorySeparatorChar}RELEASE";
+ var sparkversionstr = string.Empty;
+ if (File.Exists(releasefile))
+ {
+ // First line of the RELEASE file under SPARK_HOME will be something similar to:
+ // Spark 2.3.2 built for Hadoop 2.7.3
+ var firstLine = File.ReadLines(releasefile).FirstOrDefault();
+ var columns = firstLine?.Split(' ');
+ if (columns.Length >= 1)
+ {
+ sparkversionstr = columns[1];
+ }
+ }
+ if (string.IsNullOrWhiteSpace(sparkversionstr))
+ {
+ this.Version = new Version();
+ }
+ else
+ {
+ this.Version = new Version(sparkversionstr);
+ }
+ }
+
+ }
+}
\ No newline at end of file
From 14b6db05e1fc20587eba7141c5f6d73685922573 Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Thu, 25 Jun 2020 07:33:23 +0000
Subject: [PATCH 11/13] Fix spark home is null at unit-test
---
src/csharp/Microsoft.Spark/Utils/SparkSettings.cs | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/src/csharp/Microsoft.Spark/Utils/SparkSettings.cs b/src/csharp/Microsoft.Spark/Utils/SparkSettings.cs
index 09a8c1d06..aa5a61e4d 100644
--- a/src/csharp/Microsoft.Spark/Utils/SparkSettings.cs
+++ b/src/csharp/Microsoft.Spark/Utils/SparkSettings.cs
@@ -54,6 +54,7 @@ private void find_spark_home()
SPARK_HOME = Environment.GetEnvironmentVariable(SPARK_HOME_ENV_KEY);
if (string.IsNullOrWhiteSpace(SPARK_HOME) == false)
{
+ SPARK_HOME = string.Empty;
return;
}
foreach (var possiblehome in possible_spark_home())
@@ -89,7 +90,7 @@ private IEnumerable<string> possible_spark_home()
/// </summary>
private string locateSparkSubmit(string sparkhome)
{
- var fn = Path.Combine(sparkhome, "bin", sparksubmitcmd);
+ var fn = Path.Combine(sparkhome ?? string.Empty, "bin", sparksubmitcmd);
return File.Exists(fn)
? fn : string.Empty;
}
From 2fad72b51e08422f19d3eed0bcc26b64a787530a Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Tue, 7 Jul 2020 22:35:31 +0000
Subject: [PATCH 12/13] Fix the jvm bridge helper dispose issue
---
src/csharp/Microsoft.Spark/Sql/SparkSession.cs | 7 -------
1 file changed, 7 deletions(-)
diff --git a/src/csharp/Microsoft.Spark/Sql/SparkSession.cs b/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
index cebc1e85b..c67900a21 100644
--- a/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
+++ b/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
@@ -345,13 +345,6 @@ public UdfRegistration Udf() =>
public void Stop()
{
_jvmObject.Invoke("stop");
-
- // if we have created the jvm bridge process, dispose it now.
- if (s_jvmbridge != null)
- {
- s_jvmbridge.Dispose();
- s_jvmbridge = null;
- }
}
/// <summary>
From f7b10169170f3cdb6a2e07233f2986e9d1f7544c Mon Sep 17 00:00:00 2001
From: laneser <laneser.kuo@gmail.com>
Date: Tue, 7 Jul 2020 22:38:15 +0000
Subject: [PATCH 13/13] restore the Stop()
---
src/csharp/Microsoft.Spark/Sql/SparkSession.cs | 5 +----
1 file changed, 1 insertion(+), 4 deletions(-)
diff --git a/src/csharp/Microsoft.Spark/Sql/SparkSession.cs b/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
index c67900a21..6e9008ff5 100644
--- a/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
+++ b/src/csharp/Microsoft.Spark/Sql/SparkSession.cs
@@ -342,10 +342,7 @@ public UdfRegistration Udf() =>
/// <summary>
/// Stops the underlying SparkContext.
/// </summary>
- public void Stop()
- {
- _jvmObject.Invoke("stop");
- }
+ public void Stop() => _jvmObject.Invoke("stop");
/// <summary>
/// Returns a single column schema of the given datatype.