Spark: [QUESTION]: Streaming example takes a long time to run

[el-genius]

Describe the bug
I try to run that example
spark/examples/Microsoft.Spark.CSharp.Examples/Sql/Streaming/

good news:
it's run
bad news:
it takes much more time to represent data on the console
I didn't know what is the reason
So, if you've any expecting reason for that please share with me

[elvaliuliuliu]

@el-genius , thanks for your question. Can you please provide the command line you use and the full log?

[imback82]

@el-genius You can try with --master local[*] as part of spark-submit or you can set spark.sql.shuffle.partitions config to a low number like 3. That should speed things up.

[el-genius]

@elvaliuliuliu
this is command line
spark-submit --class org.apache.spark.deploy.dotnet.DotnetRunner --master local bin\Debug\netcoreapp3.1\microsoft-spark-2.4.x-0.10.0.jar dotnet bin\Debug\netcoreapp3.1\NetCatStream.dll

I'll share full log on Sunday morning.

[el-genius]

@imback82
I set --master local as you see in command line I used.
so, try to explain how to play with config is it setting in code?

[elvaliuliuliu]

@el-genius: You can set the config in the code or in the command like spark-submit --conf spark.sql.shuffle.partitions=3

Or try use --master local[*] instead of --master local, It will run Spark locally with as many worker threads as logical cores on your machine.

[el-genius]

good, I'll try that them inform you that make things run better or not.
thank you for your help

[el-genius]

I tried --master local[*] and spark-submit --conf spark.sql.shuffle.partitions=3
It did better but when I used the same example using python, I got the result immediately without any configuration could you tell me the reason behind that?

[imback82]

Can you share the python script? And did you run it thru pyspark or spark-submit?

[el-genius]

using pyspark the code almost the same in C# like python both using netcat by the sameway and perform the same calculation on words

[imback82]

Can you paste the script here? Thanks.

[el-genius]

import sys
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split


# Check that correct number of args have been passed as input
if __name__ == "__main__":

    if len(sys.argv) != 3:
        print("Usage: spark-submit m01_demo01_netcat.py <hostname> <port>", file=sys.stderr)
        exit(-1)


    # Extract host and port from args    
    host = sys.argv[1]
    port = int(sys.argv[2])


    # Set the app name when creating a Spark session
    # If a Spark session is already created for the app, use that. 
    # Else create a new session for that app
    spark = SparkSession\
        .builder\
        .appName("NetcatWordCount")\
        .getOrCreate()


    # Set log level. Use ERRROR to reduce the amount of output seen
    spark.sparkContext.setLogLevel("ERROR")


    # Create DataFrame representing the stream of input lines from connection to host:port
    # We're reading from the socket on the port where netcat is listening
    lines = spark\
        .readStream\
        .format('socket')\
        .option('host', host)\
        .option('port', port)\
        .load()


    # Split the lines into words
    # Explode turns each item in an array into a separate row
    # Alias sets the name of the column for the words
    # The result - each word of input is a row in a table with one column named "word"
    words = lines.select(
        explode(
            split(lines.value, ' ')
        ).alias('word')
    )


    # Generate running word count
    wordCounts = words.groupBy('word')\
                      .count()


    # Start running the query that prints the running counts to the console
    # Running in "complete" mode ensures that any operation uses ALL data 
    # - from previous and current batch 
    # The call to format sets where the stream is written to
    query = wordCounts.writeStream\
                      .outputMode('complete')\
                      .format('console')\
                      .start()

    query.awaitTermination()

[imback82]

@elvaliuliuliu Can you help repro this behavior? Thanks!

[elvaliuliuliu]

@imback82 : Sure, I will follow up on this!

@el-genius : Thanks for your question. I will take a look and update.

[elvaliuliuliu]

@el-genius : Can you provide the detailed information on how long does each language take the program to finish?

[el-genius]

@elvaliuliuliu
In C# I can write in the terminal word then another word before processing
In Python once I write the word the process finish

[el-genius]

@elvaliuliuliu
if you want I provide you with C# code also it's Ok to but I'm sure code in both C# and Python are the same

[elvaliuliuliu]

@el-genius, Can you try run Python example with spark-submit to compare?

I have run C# and Python with spark-submit, If used --master local, both C# and Python takes ~55s for each batch; If used without --master local, both C# and Python takes ~12s for me.

The behavior looks pretty much the same across these two languages. Can you please try and update here?

[el-genius]

Ok I'll try both and update ou here

[el-genius]

@elvaliuliuliu
are you tried with spark.sql.shuffle.partitions=3 or without ?

[elvaliuliuliu]

I tried without it, but as long as you run the exact same settings for both, then it will be comparable.

[elvaliuliuliu]

@el-genius : Do you have an update for the comparison as per suggestion?