Questions tagged [spark-streaming]

Spark Streaming is an extension of the core Apache Spark API that enables high-throughput, fault-tolerant stream processing of live data streams. From the version 1.3.0, it supports exactly-once processing semantics, even in face of failures.

Filter by
Sorted by
Tagged with
0 votes
0 answers
6 views

spark-streaming (DStream): Does Spark stream Support pause/resume consumption of message from Kafka on runtime

We are using Spark 3.4.0 DStream in our application for Message consumption and processing via Kafka. I have a use case, where I need to pause message consumption on some condition and resume back ...
Saurabh Agrawal's user avatar
0 votes
0 answers
13 views

SET component in ODI when using spark streaming

I am creating a mapping in ODI using spark streaming. One problem is that previously when running batch using Oracle technology, my logic had a union step and I used ODI's SET component. However, in ...
HaiCP14's user avatar
0 votes
2 answers
46 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
1 answer
29 views

In pyspark, what is the difference between dlt.read_stream() and spark.readstream()?

I am tryting to understand In pyspark, what is the difference between dlt.read_stream() and spark.readstream()? can we replace dlt.read_stream() with spark.readstream() without affecting the data? ...
Aditya Raj's user avatar
0 votes
0 answers
14 views

Azure Event Hubs: Avro v Parquet v Deltalake (preview) - most optimal output format? [closed]

I wanted to get an opinion on a design decision. I am ingesting data using Azure Event Hubs, and this initially ingest goes into a raw data layer. This is then transformed downstream. For this ingest ...
Itachi07's user avatar
0 votes
0 answers
23 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
25 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
19 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
24 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
-1 votes
1 answer
55 views

Get column rows into a string variable using Spark Dataframe

I am trying to convert column rows into a string variable. Input dataframe. depName emp_name develop Astrid develop Freja develop Wilma sales Maja sales Alice personnel John personnel Marsh ...
RMK's user avatar
  • 1
0 votes
0 answers
59 views

Converting T-SQL Procedure to PySpark Function

So there is a code in T-SQL , majorly a procedure that I want to convert to Pyspark . Here is the T-SQL procedure - CREATE PROCEDURE [dbo].[SEGMENTATION] ( @COL_NM varchar(max), @INPUT_TABLE ...
DEVEN MALI'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
0 answers
39 views

Getting some logs from kafka topic into spark and the sending it to elasticsearch

I am trying to send some data from apache kafka to elasticsearch through apache spark , for now i am testing the pipeline but still getting some errors , ultimately i will build a machine learning ...
Ilyas Ch's user avatar
0 votes
0 answers
15 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
0 votes
0 answers
35 views

jdbc postgres writes batch but not streaming in spark 3.2.4

For some reason jdbc postgresql works well for sinking batch data, but it doesn't for streaming data with my new version of spark 3.2.4, Scala 2.12.15 and hadoop 3.3.4. Jar Files: kafka-clients-3.3.2....
Ivan's user avatar
  • 119
1 vote
0 answers
29 views

Job completion tracking across kafka

I have a distributed pipeline that reads jobs from a Queue (SQS). Each job then produces millions of messages and sent over to Kafka topic TopicA to analyze. After the analysis is done, it’s then sent ...
cosmos's user avatar
  • 2,183
0 votes
1 answer
58 views

Spark dependency failed to start

I am getting a dependency failed to start error but the container is running and accessible. Also, the log file does not show any error. I can access the UI from localhost:9090. I am running docker on ...
Gregory Morris's user avatar
0 votes
1 answer
72 views

Need perform aggregation operartion in pysaprk streaming job

I'm receiving data of multiple sensors in stream in every 1 minute interval into databricks. Need to create a new sensor name 'PQRS' if sensors 'ABC' and 'DEF' is available for each pyspark streaming ...
paul saju's user avatar
-1 votes
0 answers
63 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
1 answer
27 views

Connection between an .NET application on the host and a spark-streaming application in a docker container via a socket

I need to make two applications. The .net application will send json from the host, and the spark-streaming application in the docker container will receive this data via a socket and process it. But ...
magistr777's user avatar
1 vote
1 answer
40 views

Spark Streaming output mode to process only new 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
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
1 vote
1 answer
39 views

nats-spark-connector with Java giving an error

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
0 votes
2 answers
56 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
0 answers
20 views

Count distinct values in pyspark streaming with window

I want to count distinct column values from window, but when I run query it gives error as pyspark.errors.exceptions.captured.AnalysisException: [NESTED_AGGREGATE_FUNCTION] It is not allowed to use ...
Panda's user avatar
  • 555
0 votes
0 answers
20 views

Main reason for data loss in structured spark streaming

We have deployed an application that subscribes to around 20 Kafka topics using foreachbatch and write as parquet with append using repartition and partitionBy. We have noticed that we get data loss ...
Akino76's user avatar
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
0 votes
0 answers
23 views

Spark Structured Streaming - Stream-Stream Join using sparksql

I hope everyone is doing well. I am working to create a stream-stream join Spark app and I have a few doubts about it. My sources of data is file kind and the output is to write into an Iceberg Table. ...
Vinícius Matheus Olivieri'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
1 answer
276 views

Efficiently append and update rows in delta table

I'm working with Azure databricks and needs to append and update records in a delta table. Since I'm quite new to Spark (and Databricks for that matter) my main question is if Im on the right track ...
nklsla's user avatar
  • 345
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
0 answers
80 views

Databricks autoloader checkpoint data read

In Databricks auto loader, How to read the data inside the checkpoint directory and understand For each batch how what are all the files it processed ?. I Know we can use cloud_file_state() but it's ...
Deepak Kumar's user avatar
0 votes
0 answers
75 views

Left Outer Join on Streams in Pyspark is dropping the non-matching values

I have a dlt pipeline in which i have to join 2 streams with a left outer join. and save this as a temp table (futher there will be more code that will use temp table data to make silver table). But, ...
noobCoder50'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
18 views

Spark streaming status and Blue/Green deployments

Spark streaming has a state, but how should pass between the blue/green environments? I'm on Databricks, which allows me to create clones of tables, so the new environment gets snapshots of the table ...
ozw1z5rd's user avatar
  • 3,091
1 vote
0 answers
47 views

How to use PubSubLite Spark connector in custom pyspark container for Dataproc serverless

Edit Using image version 2.0 seems to work as expected (no error). So this bug is a regression from --version 2.0 to --version 2.1 Initial message TL;DR I submit a Dataproc Serverless PubSubLite ...
AlexisBRENON's user avatar
  • 2,941
0 votes
1 answer
63 views

Pyspark java.lang.OutOfMemoryError error with wholeTextFiles

I have 1160 XML files, each of size 300MB in the data_dir. I want to count the number of whole words. I have a local machine with 256 cores and 256 GB RAM. Given that the whole data size is around ...
m0ss's user avatar
  • 338
0 votes
2 answers
47 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
0 votes
0 answers
44 views

Pyspark: Convert streaming dataframe to regular spark dataframe

I need to convert a spark streaming dataset into regular spark dataframe and then apply transformations on it in pyspark. In scala spark, the following code works: foreachBatch { (batchDf: DataFrame, ...
sks's user avatar
  • 145
1 vote
1 answer
128 views

Scala Spark Iceberg writeStream. How to set bucket?

I'm trying to write data to Iceberg table in Spark streaming (written in Scala). Writer code: val streamResult = joined.writeStream .format("iceberg") .partitionBy("...
Netrunner'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
45 views

problem reading files in delta lake house - data streaming

I'm building my first delta lake house in databricks, alone. I need to read files that are in csv format in a bucket in AWS, I can view the folder with my files by doing display(dbutils.fs.ls. However,...
Joyci Santos's user avatar
0 votes
1 answer
46 views

Handing skewed processing time of events in a streaming application

I have a streaming application (written over spark/storm/whatever does not matter). Kafka is used as a source of stream events. Now there are some events that take significantly larger resource (time, ...
best wishes's user avatar
  • 5,934
0 votes
0 answers
33 views

Condition for watermark interval for batch data in Spark Structured Streaming

I have two sources of data-itemDetails, admitDetails which come in batch mode. Sometimes this batch mode contains historical data as well. We are using Spark Structured Streaming which ingests data as ...
Rahul Kumar's user avatar
-1 votes
1 answer
161 views

How to write dataframe on s3 bucket by pyspark but without using Hadoop

I Want to write a dataframe directly on s3 bucket by pyspark but dont want to use Hadoop any how. Not single word of Hadoop required in python or pyspark code. from pyspark.sql import SparkSession ...
priyanka juyal's user avatar
0 votes
1 answer
154 views

Which time-based partitioning strategy to use for Delta or Iceberg

I am using spark-streaming to ingest live event streams every 5 minutes and append into delta or apache iceberg table. this table can be ingested and processed by downstream data pipeline or it can be ...
nir's user avatar
  • 3,771
0 votes
0 answers
15 views

Is there a way to identify whether dataloss is present while consuming data from Kafka Topic

Wanted to know is there any way to identify the dataloss from the consumer while consuming data from the Kafka Topic. Producer has sent message to the Topic but consumer is not received. For example ...
John's user avatar
  • 1
0 votes
0 answers
59 views

How to do aggregation in Spark Streaming without watermarking?

Let's assume I have a streaming data set like this: ----------------------------------------------- |package_id| timestamp | values | |----------|-----------------------|-----------| | ...
Ender's user avatar
  • 71

1
2 3 4 5
112