Questions tagged [spark-structured-streaming]
Spark Structured Streaming allows processing live data streams using DataFrame and Dataset APIs.
2,405
questions
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 = ...
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 ...
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 ...
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()
....
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 = ...
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(...
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 ...
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, ...
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", "...
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 ...
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'")
....
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 ...
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,...
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 \
...
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(&...
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....
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,...
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 ...
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....
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 = ...
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 ...
-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$...
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 ...
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&...
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 ...
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. ...
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://...
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 ...
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 ...
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. ...
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) \
....
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('...
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 ...
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,...
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 ...
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()),
...
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 ...
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 ...
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)....
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 ...
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(&...
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 .
+-----------------------+----------------------------------------------------------------------------------------...
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 ...
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 ...
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 ...
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 ...
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
...
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 ...
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 ...
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 ...