Spark Streaming with PubSub Lite

[stas-park]

I want to create a Spark Streaming application in DataProc that has a subscription to PubSub Lite or PubSub (I discovered there is no support for a regular PubSub service, so I tried to use the "Lite" service).

The idea is straightforward: as soon as a new message arrives in PubSub Lite, Spark pipeline should be processed.

Here is a simple code example I used - https://cloud.google.com/pubsub/lite/docs/samples/pubsublite-spark-streaming-from-pubsublite

from pyspark.sql import SparkSession
from pyspark.sql.types import StringType

# TODO(developer):
# project_number = 11223344556677
# location = "us-central1-a"
# subscription_id = "your-subscription-id"

spark = SparkSession.builder.appName("read-app").master("yarn").getOrCreate()

sdf = (
    spark.readStream.format("pubsublite")
    .option(
        "pubsublite.subscription",
        f"projects/{project_number}/locations/{location}/subscriptions/{subscription_id}",
    )
    .load()
)

sdf = sdf.withColumn("data", sdf.data.cast(StringType()))

query = (
    sdf.writeStream.format("console")
    .outputMode("append")
    .trigger(processingTime="1 second")
    .start()
)

# Wait 120 seconds (must be >= 60 seconds) to start receiving messages.
query.awaitTermination(120)
query.stop() 

I expect the pipeline to be triggered immediately after a new message is published into pubsub, but it doesn't happen, instead, all the accumulated messages are processed in 1 minute interval time.


1. What configuration properties am I missing?
2. Is there some 3rd party connector for a regular PubSub service or a workaround?

[ms4446 · Accepted solution]

Your current Spark Streaming setup with Pub/Sub Lite processes data every second. To enhance this for near-real-time processing, consider the following adjustments:

Reducing Trigger Interval:

• While Spark Structured Streaming doesn't support triggering on individual messages, you can approximate near-real-time processing by reducing the trigger interval.
• Example of setting it to 100 milliseconds:

# ... previous code ...
# Trigger every 100 milliseconds
query = (
    sdf.writeStream.format("console")
    .outputMode("append")
    .trigger(processingTime="100 milliseconds")
    .start()
)
# Wait 120 seconds (must be >= 60 seconds) to start receiving messages.
query.awaitTermination(120)
query.stop()

• Decreases latency but doesn't guarantee processing of each message as it arrives.
• Be aware that very short intervals can increase resource usage, such as CPU.

Exploring Pub/Sub Connectors:

• Spark doesn't have native support for Google Cloud Pub/Sub, but you can explore third-party connectors:
  • Google Cloud Dataproc Pub/Sub Connector: Official Google connector for batch processing with Dataproc, allowing Spark jobs to interact with Pub/Sub. Not designed for streaming.
  • Spark PubSub Connector: Open-source connector that integrates Spark with Google Cloud Pub/Sub. Not officially supported by Google, so evaluate its compatibility with your Spark environment.

Alternative Approaches:

• Cloud Dataflow: For real-time streaming with per-message processing, Google Cloud Dataflow is an excellent choice. Offers native Pub/Sub integration and is a fully managed service for running Apache Beam pipelines.
• Direct Pub/Sub Integration: Implement custom code within your Spark application to interact directly with the Pub/Sub API. Requires careful management of message acknowledgment and error handling.

Select the method that aligns best with your project's latency, throughput, and resource requirements.