Questions tagged [spark-structured-streaming]

Spark Structured Streaming allows processing live data streams using DataFrame and Dataset APIs.

Filter by
Sorted by
Tagged with
0 votes
1 answer
27 views

Spark Structured Streaming dropDuplicates is not working as expected

I'm  connecting to NATS Jetstream to consume the messages and process using Spark Java code. Below is the code snippet private static void sparkNatsTester() { SparkSession spark = ...
VGH's user avatar
  • 373
0 votes
0 answers
21 views

How to ensure Spark Streaming is reading new records only from selected partition in source?

I have a PySpark Streaming pipeline in Databricks which is reading from a partitioned source table, with a filter on the field the source table is partitioned on. This streaming pipeline is using the ...
Mayur Makhija's user avatar
1 vote
0 answers
22 views

After application restart, total number of processed messages at the end is not same as input messages

I'm using nats-spark-connector ( https://github.com/nats-io/nats-spark-connector/tree/main/load_balanced ) to connect to NATS Jetstream and consume the message and process using Spark Java code. Below ...
VGH's user avatar
  • 373
1 vote
0 answers
18 views

Spark Streaming - After application restart, pulled message count is not correct

I'm using below Spark Java code to pull messages from NATS Jetstream private static void sparkNatsTester() { SparkSession spark = SparkSession.builder() ....
VGH's user avatar
  • 373
0 votes
1 answer
23 views

Dynamic Filter with Spark Structured Streaming

I am working on my Spark Streaming project with the goal of creating a simple app to notify users when a data stream meets a condition (e.g., sending a notification when a stock price > x). df = ...
Baubau Tran's user avatar
0 votes
1 answer
39 views

Efficient reading from Kafka with Spark Streaming

I have an app, which gets data from Kafka and save it to the database. My code looks like this: spark.readStream .format("kafka") .options(options) .load() .writeStream .trigger(...
Gar Garrison's user avatar
0 votes
0 answers
47 views

Spark Streaming with Kafka Problem: AnalysisException: Failed to find data source: kafka

I have an issue while using Spark Streaming and Kafka: I am running Spark 3.5.0 and Kafka 3.5 on my virtual machine (VMWare), and I want to read messages from my laptop with Spark Streaming (which is ...
Amatuer's user avatar
0 votes
2 answers
79 views

How exactly does dropDuplicatesWithinWatermark work?

I have this code: import org.apache.spark.sql.SparkSession import org.apache.spark.sql.execution.streaming.MemoryStream import org.apache.spark.sql.streaming.{OutputMode, StreamingQueryListener, ...
pavel_orekhov's user avatar
0 votes
0 answers
50 views

Pyspark streaming how to restructure JSON File

this is my input data stream for kafka i want to restructure in pyspark streaming to be in a JSON format link the one in the output . input stream: {"name": "position_y", "...
Waleed saeed's user avatar
0 votes
1 answer
21 views

Multiple CSV Scan in spark structured streaming for same file in s3

I have a spark 3.4.1 structured streaming job which uses custom s3-sqs connector by AWS which reads messages from sqs, reads the provided path in SQS Message and then read from S3. Now I need to split ...
Umang Desai's user avatar
0 votes
0 answers
15 views

Joining aggregate streaming dataframe with streaming dataframe in PySpark

Here is the code I'm trying to run streaming_df = ( spark.readStream .option("option1", "value1") ) agg_df = ( streaming_df .filter("filter1 = 'value1'") ....
Ravi Kiran.M's user avatar
0 votes
0 answers
49 views

Streaming data from apache kafka to apache spark and then to elasticsearch in ubuntu

I want to read data from an apache kafka topic through apache spark with a python script and then perform some operations on the data and then stream it to elasticsearch in real time ; ultimately i ...
Ilyas Ch's user avatar
0 votes
0 answers
14 views

Dynamic Spark Structured Streaming: Handling Stream-Stream Joins with Changing

I want to create a simple application using Spark Structured Streaming to alert users (via email, SMS, etc.) when stock price data meets certain requirements. I have a data stream: data_stream However,...
Baubau Tran's user avatar
4 votes
2 answers
162 views

Databricks Autoloader Schema Evolution throws StateSchemaNotCompatible exception

I am trying to use Databricks Autoloader for a very simple use case: Reading JSONs from S3 and loading them into a delta table, with schema inference and evolution. This is my code: self.spark \ ...
Robert Kossendey's user avatar
0 votes
0 answers
44 views

I have a problem when Spark connects Kafka

Here is my code: from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName('Spark') \ .getOrCreate() df = spark.readStream \ .format(&...
Thanh Tùng Lê's user avatar
0 votes
0 answers
30 views

Is there a way to cache or reuse PythonUDFRunner environment to reduce initialization overhead in Pyspark structured streaming

I am working on a Pyspark structured streaming application to invoke an UDF functionality on every event. from pyspark.sql import SparkSession from pyspark.sql.functions import * from pyspark.sql....
KMK's user avatar
  • 1
1 vote
0 answers
30 views

Stock Price Alert Application Using Spark Structured Streaming

I am currently working on my Big Data class project, focusing on building a stock price alerting app (via email) using Spark Streaming and Kafka. Sample stock trading data stockName,tradeType,price,...
Baubau Tran's user avatar
0 votes
0 answers
22 views

How to use spark structured stream with sliding window and trigger once

I would like to use spark structured streaming to process input data and create features from it. Some features will be based on a rolling window with a stride smaller than the window, so the same ...
vincentp's user avatar
0 votes
2 answers
33 views

Pyspark: AnalysisException: Queries with streaming sources must be executed with writeStream.start();

I have multiple functions that I want to run but I get an error that says I need to use .start(). Can someone show me how to proceed? Here is a part of my code: def parse_data(df): cachedDf = df....
Alexis Kan's user avatar
0 votes
1 answer
41 views

How do I gracefully shutdown a StructureStreaming use PySpark?

as topic, how do I gracefully shutdown a StructureStreaming use PySpark? when I call spark.streams.awaitAnyTermination(timeout=100) it always return error:ERROR MicroBatchExecution: Query [id = ...
BaronChen's user avatar
0 votes
0 answers
44 views

Autoloader limitation [closed]

As per the Autoloader document in databricks, that infers schema only for the first 50 GB data or 1000 files that it discovers. So how can we tackle if the frequency of getting updates of the existing ...
Shekhar Gupta's user avatar
-1 votes
0 answers
62 views

An error occurred while calling o36.load. : java.lang.NoClassDefFoundError: scala/$less$colon$less

When I want to read the data from kafka to spark WARNING:root:kafka dataframe could not be created because: An error occurred while calling o36.load. : java.lang.NoClassDefFoundError: scala/$less$...
Helen's user avatar
  • 1
0 votes
0 answers
24 views

Circuitbreaker object inside a class on a spark executor

I have a Kafka Structured Streaming Application, and a class with a method that is run on every record inside a mapPartition. This method contains a call to Redis (Lettuce Client) that I want to add ...
Swathi Mudda's user avatar
1 vote
1 answer
81 views

What is saved by spark Autoloader checkpoint?

I have the following code I wish to use for copying data arriving in a folder on a regular basis: val streamingQuery = spark.readStream .format("cloudFiles") .option("cloudFiles.format&...
Tamás Godányi's user avatar
0 votes
0 answers
33 views

Filtering erroneous record from spark structured streaming dataframe while writing to cassandra

I know that there is some issue in nth row of my spark Scala dataframe(let's say data type is not proper). When I try to write this dataframe in cassandra using spark structured streaming, it fails ...
Aman Kumar's user avatar
0 votes
2 answers
55 views

Spark stream leftOuter not working with third stream join

I am stuck with this stream leftOuter join. I am able to stream join 2 data frames and can get null values after watermark time expires. But I am not able to achieve if I join with 3 data frames. ...
Paulnaveen's user avatar
0 votes
0 answers
38 views

Time window aggregation in separate streams followed by stream-stream join not returning results (Spark 3.5.0)

I'm trying to reproduce what is called Time window aggregation in two streams followed by stream-stream join with event time window and that was added to Spark documentation in Spark 3.5.0: https://...
azera's user avatar
  • 68
0 votes
1 answer
65 views

Non-time-based windows are not supported on streaming DataFrames/Datasets, How to resolve this?

In my code there is a specific requirement due to which i had to use partition by on a streaming dataframe. But gives me following error Non-time-based windows are not supported on streaming ...
noobCoder50's user avatar
1 vote
1 answer
80 views

Autoloader schema evolution using foreachBatch

Im faicing some schema evolution in my work process and I cant find the away to make it work. Last week, I had 5 columns enabled in the ERP system and business requried to add the 6th column in the ...
Greencolor's user avatar
0 votes
0 answers
9 views

Device disconnection(Time out) logic in pyspark stream

I'm trying if kafka doesn't receive any message for 45 seconds query running should be stopped(query.stop()), In middle of that 45 seconds if new message received start time needs to be reset again. ...
Ajith Lakshmanan's user avatar
0 votes
0 answers
28 views

I am using Kafka streaming in azure databricks using foreachbatch but checkpoint is not updating

Why checkpoint is not updating? df2.writeStream \ .foreachBatch(lambda batch_df, batch_id: processBatch(batch_df, batch_id)) \ .option("checkpointLocation", checkpoint) \ ....
Prashanth Bethi's user avatar
0 votes
0 answers
12 views

Spark streaming from a table, how to recover

Let's consider a streaming process that reads from a table: df_vbronze_google_similar_apps = spark.readStream.table('${var.database}.bronze_google_similar_apps').\ createOrReplaceTempView('...
ozw1z5rd's user avatar
  • 3,091
0 votes
0 answers
32 views

Parsing xml using Kafka + Spark Streaming to Oracle database

I am using spark to read streaming data from a kafka topic (messages are xml records), then convert the parsing into a dataframe so that it can be inserted into an Oracle database table as a stream. I ...
HaiCP14's user avatar
0 votes
0 answers
14 views

Spark streaming LOCAL run doesn't start when doing foreachbatch or flatMapGroupWithState

I'm facing a strange issue, I'm doing some tests about structured streaming, to do that I'm creating some main executed in LOCAL mode, everything was working fine until I tested flatMapGroupsWithState,...
D. belvedere's user avatar
0 votes
1 answer
68 views

How can we Force Databricks Structured Streaming to read data from event hubs every few seconds even if there is no new data?

We are using Databricks structured streaming to read data from azure event hubs and we are using forEachBatch to upsert the data to a delta table in the writeStream part The issue is - the events are ...
azuresnowflake1's user avatar
0 votes
1 answer
66 views

PySpark Dataframe Transformation from_json

i have below JSON which i am reading from Kafka and then tyring to convert to StructType using from_json function. schema_session_start = StructType([ StructField("ID", StringType()), ...
Virendar Kumar's user avatar
1 vote
1 answer
54 views

Does stage kill from Spark UI causes a reprocessing of the data?

In Spark UI there is this capability of killing an actively running stage: When a stage is killed with this button, the tasks that are associated with this stage will be reprocessed? Or they'll be ...
Cosmin Ioniță's user avatar
0 votes
1 answer
47 views

Exception in Pyspark Structured Streaming from Kafka Source

Environment: Spark 3.5.0, Scala 2.12.18, openjdk 11.0.20.1 I'm trying to stream data from a Kafka source, but getting an exception from Spark. It seems like this is often due to dependency version ...
T.Line's user avatar
  • 1
0 votes
0 answers
36 views

Spark Structured Stream not loading csv files properly

I have pipe-delimited csv files in s3 that I'm trying to load in Databricks. When reading with following code below, the file is read correctly: df_test = spark.read.option("header", True)....
almosthavoc's user avatar
0 votes
0 answers
71 views

Spark-Kafka Integration not working: Kafka broker with producer and consumer script are getting stuck as soon as we run spark script(consumer.py)

I'm trying to read data from kafka topic by using spark structured streaming on ec2(ubuntu) machine. If I try to read the data by using kafka stream only(kafka-console-consumer.sh) then there is no ...
Rushi's user avatar
  • 23
0 votes
1 answer
35 views

Problem with expanding json data in a column into multiple columns in a dataframe PySpark from Apache Kafka streaming data

Dears, I practise streaming data. I decided to create Apache Kafka. I would like to read data from KafkaProducer, then I created a Kafka Source for Streaming Queries: df = spark \ .read \ .format(&...
lukasz93's user avatar
0 votes
2 answers
44 views

PySpark Dataframe Transformation pyspark

I have a below dataframe which i need to transform as below. i am using PySpark 3.4.1 . +-----------------------+----------------------------------------------------------------------------------------...
Virendar Kumar's user avatar
1 vote
1 answer
53 views

PySpark (Spark v3.4.1) structured streaming how to implement cumulative aggregated data to write into spark sink?

I am working in pyspark structured streaming programing to generate some cumulative aggregation of streaming data. I have used mongodb and kafka as spark read stream data. I have tried foreachBatch ...
Subhadip Sahoo's user avatar
0 votes
0 answers
39 views

How to decrease CPU Usage in Spark App (python)

I've built spark app in python3 with this architecture: 1.ws_server --> json_data 2.json_data--> spark app(socket streaming) 3.spark app(socket streaming) --> POST & PUT requests to get ...
Дмитрий Григорчук's user avatar
1 vote
0 answers
36 views

How to convert a sequence into a streaming DataFrame?

Having a sequence of strings Seq[String], how can I convert it into a streaming DataFrame? The functions SparkSession.createDataFrame(...) create static DataFrames only. And the class MemoryStream ...
pgrandjean's user avatar
0 votes
1 answer
31 views

Structured Streaming + Kafka Integration numInputRows is 0 but offset is increase

Structured Streaming + Kafka Integration ,spark version 3.3 and I found something strange, in the begining I can get right numInputRows ,but about 5 times fetch numInputRows would be 0 ,bue endOffset ...
liush's user avatar
  • 13
0 votes
1 answer
56 views

use-case streaming in pyspark

I am working with Databricks on Azure, my data are hosted on ADLS2. Current version of runtime is 10.4 LTS (I can upgrade if needed) I have a table Pimproduct: id name action dlk_last_modified ...
Steven's user avatar
  • 14.4k
0 votes
1 answer
52 views

How to extract schema id from avro message in Spark Scala

I have spark Scala dataframe with column contains the value of avro message(Array[Byte]). I know that the 0 byte is the magic byte and the bytes in positions 1-4 included is the schema id. how can i ...
Tal's user avatar
  • 71
1 vote
0 answers
80 views

Databricks - Spark streaming auto-loader wildcard

Need som help with an issue loading a subdirectory from S3 bucket using auto-loader. For example: S3://path1/path2/databases*/paths/ In databases there are various versions of databases. For example ...
Josh Gomez's user avatar
0 votes
1 answer
52 views

How to read .csv and .csv.gz file with all executors in EMR cluster

I am using aws emr service (emr 6.9.0) with one core node and 8 executors of instance type m5.2xlarge. My requirement is to execute read csv or csv.gz file through a spark scala job and write it into ...
Shan's user avatar
  • 31

1
2 3 4 5
49