org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start() explained

September 17, 2017 • Apache Spark Structured Streaming • Bartosz Konieczny

Looking for a book that defines and solves most common data engineering problems? I'm currently writing one on that topic and the first chapters are already available in 👉 Early Release on the O'Reilly platform

I also help solve your data engineering problems 👉 [email protected] 📩

The first section shows the case producing this error. The second section explains why the exception occurs. Finally, the third section gives the solution.

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

Producing this exception is quite easy. Let's create a streaming with structured streaming and apply the consumption logic of low level APIs (e.g. foreach method):

Note that the code above is perfectly valid in Spark SQL semantics:

However, the first snippet executed in the context of structured streaming gives the following exception:

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

To understand the problem, we can do several things. In our case, we'll jump between exception's stack trace execution lines and discover what every method is doing.

First of all, the check producing the exception starts in org.apache.spark.sql.execution.QueryExecution#assertSupported() . This method is triggered only when the configuration entry spark.sql.streaming.unsupportedOperationCheck is set to true, what is the case of executed code.

assertSupported() calls utility method UnsupportedOperationChecker#checkForBatch(plan: LogicalPlan) that executes the following function in logic plan (children + parent):

As you can see, if one of plan's parts is a streaming, the exception we've met is thrown. The key to understand why is hidden in this famous isStreaming part. This isStreaming is in fact a method checking if any of logical plan's parts has a streaming data source:

But what is the difference between streaming and batch data sources ? Quite easy - streaming data source, unlike the batch one, returns data in continuous manner. And how Spark can distinguish them ? In fact, isStreaming method is defined in LogicalPlan abstract class and the implementation quoted above is only the default one . The logical plan is implemented by every operation leaf (algebraic or language construct) in Spark SQL, as DISTINCT (org.apache.spark.sql.catalyst.plans.logical.Distinct), JOIN (org.apache.spark.sql.catalyst.plans.logical.Join) or WHERE (org.apache.spark.sql.catalyst.plans.logical.Filter). But it's also implemented by structured streaming parts, as org.apache.spark.sql.execution.streaming.StreamingRelation used to make a link between data source (Kafka in our case) and logical plan. And this short implementation of LogicalPlan looks like:

As you see, StreamingRelation overrides isStreaming method and returns true. And because this logical plan implementation is constructed when the DataStreamReader.load() is called, the code written with batch logic can't simply fit to the streaming requirements.

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

Since the problem is our batch-oriented code, nothing more simple to fix it than converting the code to streaming-oriented. In structured streaming Spark programmers introduced the concepts of sources and sinks. The second ones define how the data is consumed. There are several basic sinks, as foreach (data read in foreach loop), console (data printed to the console) or file (data persisted to files).

The sinks are created directly from read stream by calling writeStream() method on loaded DataFrame. It will create an instance of DataStreamWriter class that can be used to consume streamed data just after calling its start() method.

Fixed code consuming data in foreach loop looks like that:

At first glance the use of structured streaming is not obvious. It differs a little from the DStream-based streaming. The first meaningful difference is that we can't easily consume read data by directly calling actions on source stream. Instead of that we must define the sink to which retrieved data will be put.

TAGS: #Kafka integration #Kafka Spark Structured Streaming

Read also about org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start() explained here:

If you liked it, you should read:

📚 Newsletter Get new posts, recommended reading and other exclusive information every week. SPAM free - no 3rd party ads, only the information about waitingforcode! Subscribe