Spark Streaming

[AshaSalorina]

Hello, I tried to find about how to use spark streaming in this project but got nothing.(not structured streaming)
Even I tried "StreamingContext" and nothing be found.
So I want to ask if we have this function or not.
Sorry for my bad english...

[imback82]

We don't currently support Spark Streaming. Any reason the Structured Streaming doesn't work for your scenario?

[AshaSalorina]

Our homework need Spark Streaming to monitor a directory.
Structured Streaming is good but I can't use that to solve this problem.
Could you please give me some suggestion for how to monitor a directory and deal them with spark?

[rapoth]

@AshaSalorina: Spark Structured streaming can be used to monitor a directory as well. Have you tried it already and it is not working?

Here's an example Spark Structured Streaming example in .NET that reads data from a directory (and everytime you drop a new file, it picks it up i.e., it also monitors):

using Microsoft.Spark.Sql;

namespace SparkStreaming
{
    class Program
    {
        static void Main(string[] args)
        {
            // Instantiate a session
            var spark = SparkSession.Builder().GetOrCreate();

            // Specify the directory on your machine where files are getting generated
            var csvDF = spark
                            .ReadStream()
                            .Option("sep", ";")
                            .Schema("name string, age integer")
                            .Csv(@"c:\tmp\data\");

            // Setup your query
            var query = csvDF
                            .WriteStream()
                            .Format("console")
                            .Option("truncate", "false")
                            .Start();

            // Wait forever
            query.AwaitTermination();
        }
    }
}

Starting the job (assumes .NET Framework 4.6.1):

spark-submit.cmd --class org.apache.spark.deploy.DotnetRunner --master local .\microsoft-spark-2.3.x-0.3.0.jar .\SparkStreaming.exe

Let me know if you run into any problems.

[AshaSalorina]

@AshaSalorina: Spark Structured streaming can be used to monitor a directory as well. Have you tried it already and it is not working?

Here's an example Spark Structured Streaming example in .NET that reads data from a directory (and everytime you drop a new file, it picks it up i.e., it also monitors):

using Microsoft.Spark.Sql;

namespace SparkStreaming
{
    class Program
    {
        static void Main(string[] args)
        {
            // Instantiate a session
            var spark = SparkSession.Builder().GetOrCreate();

            // Specify the directory on your machine where files are getting generated
            var csvDF = spark
                            .ReadStream()
                            .Option("sep", ";")
                            .Schema("name string, age integer")
                            .Csv(@"c:\tmp\data\");

            // Setup your query
            var query = csvDF
                            .WriteStream()
                            .Format("console")
                            .Option("truncate", "false")
                            .Start();

            // Wait forever
            query.AwaitTermination();
        }
    }
}

Starting the job (assumes .NET Framework 4.6.1):

spark-submit.cmd --class org.apache.spark.deploy.DotnetRunner --master local .\microsoft-spark-2.3.x-0.3.0.jar .\SparkStreaming.exe

Let me know if you run into any problems.

Hello and thanks for your help.It's really helpful to me.
I tried this recently.It works well but there still have some problems.
Because of no “foreach”, I can not do some thing like filter them in different way(when I tried use sql like select or where, each time the streaming need “writeStream” to end).Even I can't use Collect() to do something like 'foreach(var item in streamingDF.Collect())' what work with common df.It needs end when whatever I wanna do.I can't use rows twice.
So, did you have any plans about add a foreach support?
By the way, did I can add a new row to DF like df.loc() in python?I can't find any way to do that or I miss some functions?

[AshaSalorina]

An other problem:

java.lang.UnsupportedOperationException: Data source jdbc does not support streamed writing

sad ><

[imback82]

An other problem:

java.lang.UnsupportedOperationException: Data source jdbc does not support streamed writing

Did you try examples from here: https://databricks.com/blog/2016/07/28/structured-streaming-in-apache-spark.html ?

[imback82]

So, did you have any plans about add a foreach support?

No, but can you try using UDF instead?