I am building a Spark Structured Streaming application where I am doing a batch-stream join. And the source for the batch data gets updated periodically. So, I am planning to do a persist/unpersist...

I want to write one large sized dataframe with repartition, so I want to calculate number of repartition for my source dataframe. numberofpartition = {size of dataframe/default_blocksize} How to c...

I have a Spark Streaming job which has been running continuously. How do I stop the job gracefully? I have read the usual recommendations of attaching a shutdown hook in the job monitoring and send...

I am trying to write data on an S3 bucket from my local computer: spark = SparkSession.builder \ .appName('application') \ .config("spark.hadoop.fs.s3a.access.key", configuration.AWS_AC...

I need to know the fileName for the input file that is streamed from the input dir. Below is the spark FileStreaming code in scala programming object FileStreamExample { def main(args: Array[St...

I keep getting the the following exception very frequently and I wonder why this is happening? After researching I found I could do .set("spark.submit.deployMode", "nio"); but that did not work eit...

I have a Spark Streaming application that reads data from multiple Kafka topics. Each topic has a different type of data, and thus requires a different processing pipeline. My initial solution was...

I am using spark 1.5.2. I need to run spark streaming job with kafka as the streaming source. I need to read from multiple topics within kafka and process each topic differently. Is it a good idea...

I'm facing an issue related to Kafka. I'm having my current service (Producer) that sends the message to a Kafka topic (events). The service is using kafka_2.12 v1.0.0, written in Java. I'm tryin...

When I run the following test, it throws "Cannot call methods on a stopped SparkContext". The possible problem is that I use TestSuiteBase and Streaming Spark Context. At the line val gridEvalsRDD ...

I need to modify a class by adding two new parameters. This class is serialized with Kryo. I'm currently persisting the information related to this class, among other things, as an RDD, every time ...

I am trying to read a kafka stream and save it to Hive as a table. The consumer code is : import org.apache.spark.sql.{DataFrame, Dataset, SaveMode, SparkSession} import org.apache.spark.sql.functi...

I have a simple Spark application running on cluster mode. val funcGSSNFilterHeader = (x: String) => { println(!x.contains("servedMSISDN") !x.contains("servedMSISDN") } val ssc = new Stream...

I have a scenario in my project , where I am reading the kafka topic messages using spark-sql-2.4.1 version. I am able to process the day using structured streaming. Once the data is received and a...

I'm processing events using Dataframes converted from a stream of JSON events which eventually gets written out as Parquet format. However, some of the JSON events contains spaces in the keys which...

I am looking to see if there is something like AWS Glue "bookmark" in spark. I know there is checkpoint in spark which works well on individual data source. In Glue we could use bookmark ...

I am trying to submit spark-submit but its failing with as weird message. Error: Could not find or load main class org.apache.spark.launcher.Main /opt/spark/bin/spark-class: line 96: CMD: bad arr...

I am using Spark 1.3.0 with python api. While transforming huge dataframes, I cache many DFs for faster execution; df1.cache() df2.cache() Once use of certain dataframe is over and is no longer ...

Trying to create a test for spark data streaming writeStream function as shown below: SparkSession spark = SparkSession.builder().master("local").appName("spark session").getOrCreate() val lakeD...

I have a simple Spark job that streams data to a Delta table. The table is pretty small and is not partitioned. A lot of small parquet files are created. As recommended in the documentation (https:...

I'm doing the window-based sorting for the Spark Structured Streaming: val filterWindow: WindowSpec = Window .partitionBy("key") .orderBy($"time") controlDataFrame=controlDat...

I know there are many threads already on 'spark streaming connection refused' issues. But most of these are in Linux or at least pointing to HDFS. I am running this on my local laptop with Windows....

In my spark job, I tried to overwrite a table in each microbatch of structured streaming batchDF.write.mode(SaveMode.Overwrite).saveAsTable("mytable") It generated the following error. ...

Given the following series of events: df1 = read df2 = df1.action df3 = df1.action df2a = df2.action df2b = df2.action df3a = df3.action df3b = df3.action df4 = union(df2a, df2b, df3a, d3b) df4.col...

I am monitoring a spark executor JVM of a OutOfMemoryException. I used Jconsole to connect to executor JVM. Following is the snapshot of Jconsole: In the image used memory is shown as 3.8G and co...