Expose DataStreamWriter.ForeachBatch API #549

suhsteve · 2020-06-17T05:46:35Z

This PR exposes the DataStreamWriter.ForeachBatch API

Users can call this API by using an Action<DataFrame, long> delegate. For example

spark
    .ReadStream()
    .Schema("id INT")
    .Csv("/path/to/csv")
    .WriteStream()
    .ForeachBatch((df, id) => 
    {
        df.Write().Csv("/unique/write/path"));
    }

Logs produced on the dotnet side:

[2020-06-21T20:55:22.0427094Z] [Info] [CallbackServer] Starting CallbackServer.
[2020-06-21T20:55:22.0581917Z] [Info] [CallbackServer] Started CallbackServer on 127.0.0.1:61344
[2020-06-21T20:55:26.1706082Z] [Info] [CallbackConnection] [1] Connected with RemoteEndPoint: 127.0.0.1:61350
[2020-06-21T20:55:26.1754979Z] [Info] [CallbackServer] Pool snapshot: [NumThreads:1], [NumConnections:1]
[2020-06-21T20:55:26.1779739Z] [Info] [CallbackConnection] [1] Received request for callback id: 1, callback handler: Microsoft.Spark.Interop.Ipc.ForeachBatchCallbackHandler
[2020-06-21T20:55:28.7991929Z] [Debug] [CallbackConnection] [1] Received END_OF_STREAM signal.
[2020-06-21T20:55:29.9014111Z] [Info] [CallbackConnection] [1] Received request for callback id: 1, callback handler: Microsoft.Spark.Interop.Ipc.ForeachBatchCallbackHandler
[2020-06-21T20:55:30.6256825Z] [Debug] [CallbackConnection] [1] Received END_OF_STREAM signal.

Logs produced on the JVM side:

20/06/21 13:55:22 INFO DotnetBackendHandler: Connecting to a callback server at 127.0.0.1:61344
20/06/21 13:55:26 INFO CallbackConnection: Calling callback [callback id = 1] ...
20/06/21 13:55:26 INFO CallbackConnection: Signaling END_OF_STREAM.
20/06/21 13:55:28 INFO CallbackConnection: Received END_OF_STREAM signal. Calling callback [callback id = 1] successful.
20/06/21 13:55:29 INFO CallbackConnection: Calling callback [callback id = 1] ...
20/06/21 13:55:29 INFO CallbackConnection: Signaling END_OF_STREAM.
20/06/21 13:55:30 INFO CallbackConnection: Received END_OF_STREAM signal. Calling callback [callback id = 1] successful.
20/06/21 13:55:32 INFO DotnetBackendHandler: Requesting to close callback client
20/06/21 13:55:32 INFO CallbackClient: Shutting down.

suhsteve · 2020-06-17T05:47:40Z

microsoft-spark-2.3.x and 3.0.x files will be added/updated after we finalize the review for 2.4.x

src/csharp/Microsoft.Spark/Interop/Ipc/CallbackServer.cs

imback82 · 2020-06-17T18:30:25Z

.../microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackendHandler.scala

@@ -67,32 +66,19 @@ class DotnetBackendHandler(server: DotnetBackend)
              writeInt(dos, -1)
          }
        case "connectCallback" =>
-          val t = readObjectType(dis)
+          var t = readObjectType(dis)


let's make it simpler and allow only one call to "connectCallback"

src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/Streaming/DataStreamWriterTests.cs

src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackClient.scala

...la/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackConnection.scala

suhsteve · 2020-06-18T07:21:19Z

src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/Streaming/DataStreamWriterTests.cs

@@ -86,13 +88,17 @@ public void TestForeachBatch()
                .WriteStream()
                .ForeachBatch((df, id) =>
                {
-                    df.Write().Csv(Path.Combine(dstTempDirectory.Path, id.ToString()));
+                    Func<Column, Column> innerUdf = Udf<int, int>(i => i + 200);


This won't work until the bug re broadcast variables is fixed.

Can you explain why this is related to the broadcast variable bug?

s_jvmBroadcastVariables is [ThreadStatic] and is null when the ForeachBatchCallbackHandler is called (since it runs in a separate worker task.)

oh ok. it's in a critical path (CreatePythonFunction). :)

src/csharp/Microsoft.Spark/Interop/Ipc/CallbackConnection.cs

- Start CallbackServer and Worker as threads instead of tasks.

imback82

Early feedback (4 C# files remaining)

src/csharp/Microsoft.Spark/Interop/SparkEnvironment.cs

src/csharp/Microsoft.Spark/Interop/Ipc/ICallbackHandler.cs

src/csharp/Microsoft.Spark/Interop/Ipc/ForeachBatchCallbackHandler.cs

imback82 · 2020-06-21T00:35:45Z

src/csharp/Microsoft.Spark.UnitTest/TestUtils/MakeConsoleWork.cs

+    // Tests can subclass this to get Console output to display when using
+    // xUnit testing framework.
+    // Workaround found at https://github.com/microsoft/vstest/issues/799
+    public class MakeConsoleWork : IDisposable


Can you rename this class please?

.../microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackendHandler.scala

src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackClient.scala

...la/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackConnection.scala

imback82

Reviewed all files now. Looking really good.

.../microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackendHandler.scala

src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala

...la/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackConnection.scala

src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackClient.scala

src/csharp/Microsoft.Spark/Interop/Ipc/ForeachBatchCallbackHandler.cs

src/csharp/Microsoft.Spark/Interop/Ipc/CallbackServer.cs

src/csharp/Microsoft.Spark/Interop/Ipc/CallbackConnection.cs

src/csharp/Microsoft.Spark/Interop/Ipc/CallbackServer.cs

src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala

...la/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackConnection.scala

imback82

LGTM, thanks @suhsteve for working so hard on this!

suhsteve · 2020-06-23T05:42:20Z

@imback82 thanks for reviewing! Let me add the relevant files to microsoft-spark-2.3.x and 3.0.x

suhsteve added 2 commits June 16, 2020 22:37

initial commit

8a5cc90

Merge branch 'master' into foreachbatch

1fde50b

suhsteve added 3 commits June 16, 2020 23:11

minor cleanup.

630e767

change access modifier.

5cfd8c1

remove unused imports.

8d2d07e

imback82 reviewed Jun 17, 2020

View reviewed changes

src/csharp/Microsoft.Spark/Interop/Ipc/CallbackServer.cs Show resolved Hide resolved

imback82 reviewed Jun 17, 2020

View reviewed changes

src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/Streaming/DataStreamWriterTests.cs Outdated Show resolved Hide resolved

imback82 reviewed Jun 17, 2020

View reviewed changes

src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackClient.scala Outdated Show resolved Hide resolved

imback82 reviewed Jun 17, 2020

View reviewed changes

...la/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/CallbackConnection.scala Outdated Show resolved Hide resolved

PR comments.

c47a0db

suhsteve commented Jun 18, 2020

View reviewed changes

src/csharp/Microsoft.Spark/Interop/Ipc/CallbackConnection.cs Outdated Show resolved Hide resolved

suhsteve added 11 commits June 18, 2020 22:19

- Add CallbackServerTests

c09d3dc

- Start CallbackServer and Worker as threads instead of tasks.

Test scenario where ICallbackHandler does and doesn't return a value.

a081501

Test RegisterCallback

d8afed4

Rename test.

8ed0d83

Rename test.

06006a7

rename test handlers classes.

0b9e8b1

rename

10e561a

Add CallbackConnection tests.

5c03fd5

rename handler.

4b2aa27

fix comment.

8e76cc1

update exception message.

8e7a02a

imback82 reviewed Jun 21, 2020

View reviewed changes

suhsteve added 2 commits June 21, 2020 00:00

PR comments.

fc66d2d

start callback threads as background thread.

8abca76

suhsteve force-pushed the foreachbatch branch from 29dfa75 to 8abca76 Compare June 21, 2020 07:22

remove callback ability to return values.

ad401d7

suhsteve added 2 commits June 21, 2020 03:08

update test

0850505

PR comment

69976ee

imback82 reviewed Jun 22, 2020

View reviewed changes

suhsteve added 3 commits June 21, 2020 21:33

PR comments.

bce3f38

Add size of callback data to stream.

fd2d5c8

clean up test.

2f1fa2c

imback82 reviewed Jun 23, 2020

View reviewed changes

PR comments.

95fe7db

suhsteve force-pushed the foreachbatch branch from a091c51 to 95fe7db Compare June 23, 2020 05:33

Merge branch 'master' into foreachbatch

8df363e

imback82 previously approved these changes Jun 23, 2020

View reviewed changes

add scala files to other microsoft-spark versions.

bf8708a

suhsteve dismissed imback82’s stale review via bf8708a June 23, 2020 06:01

imback82 approved these changes Jun 23, 2020

View reviewed changes

imback82 merged commit 29ad2cb into dotnet:master Jun 23, 2020

suhsteve deleted the foreachbatch branch September 6, 2020 05:52

imback82 mentioned this pull request Sep 18, 2020

Broadcast encryption support #489

Merged

Expose DataStreamWriter.ForeachBatch API #549

Expose DataStreamWriter.ForeachBatch API #549

Uh oh!

Conversation

suhsteve commented Jun 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

suhsteve commented Jun 17, 2020

Uh oh!

Uh oh!

imback82 Jun 17, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

suhsteve Jun 18, 2020

Choose a reason for hiding this comment

Uh oh!

imback82 Jun 18, 2020

Choose a reason for hiding this comment

Uh oh!

suhsteve Jun 18, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

imback82 Jun 18, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

imback82 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

imback82 Jun 21, 2020

Choose a reason for hiding this comment

Uh oh!

suhsteve Jun 21, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

imback82 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

imback82 left a comment

Choose a reason for hiding this comment

Uh oh!

suhsteve commented Jun 23, 2020

Uh oh!

Uh oh!

suhsteve commented Jun 17, 2020 •

edited

Loading

suhsteve Jun 18, 2020 •

edited

Loading