Using Pub/Sub Lite with Apache Spark

The Pub/Sub Lite Spark Connector is an open-source Java client library that supports the use of Pub/Sub Lite as an input and output source for Apache Spark Structured Streaming . The connector works in all Apache Spark distributions, including Dataproc.

This quickstart shows you how to:

  • read messages from Pub/Sub Lite
  • write messages to Pub/Sub Lite

using PySpark from a Dataproc Spark cluster.

Before you begin

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
  2. In the Google Cloud Console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. Make sure that billing is enabled for your Cloud project. Learn how to confirm that billing is enabled for your project.

  4. Enable the Pub/Sub Lite, Dataproc, Cloud Storage, Logging APIs.

    Enable the APIs

  5. Install and initialize the Cloud SDK.
  6. In the Google Cloud Console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  7. Make sure that billing is enabled for your Cloud project. Learn how to confirm that billing is enabled for your project.

  8. Enable the Pub/Sub Lite, Dataproc, Cloud Storage, Logging APIs.

    Enable the APIs

  9. Install and initialize the Cloud SDK.

Set up

  1. Create variables for your project.

    export PROJECT_ID=$(gcloud config get-value project)
    export PROJECT_NUMBER=$(gcloud projects list \
        --filter="projectId:$PROJECT_ID" \
        --format="value(PROJECT_NUMBER)")
    
  2. Create a Cloud Storage bucket. Cloud Storage bucket names must be globally unique.

    export BUCKET=your-bucket-name
    gsutil mb gs://$BUCKET
    
  3. Create a Pub/Sub Lite topic and subscription in a supported location.

    export TOPIC=your-lite-topic-id
    export SUBSCRIPTION=your-lite-subscription-id
    export PUBSUBLITE_LOCATION=your-lite-location
    gcloud pubsub lite-topics create $TOPIC \
        --location=$PUBSUBLITE_LOCATION \
        --partitions=2 \
        --per-partition-bytes=30GiB
    gcloud pubsub lite-subscriptions create $SUBSCRIPTION \
        --location=$PUBSUBLITE_LOCATION \
        --topic=$TOPIC
    
  4. Create a Dataproc cluster.

    export DATAPROC_REGION=your-dataproc-region
    export CLUSTER_ID=your-dataproc-cluster-id
    gcloud dataproc clusters create $CLUSTER_ID \
       --region $DATAPROC_REGION \
       --image-version 1.5 \
       --scopes 'https://www.googleapis.com/auth/cloud-platform' \
       --enable-component-gateway \
       --bucket $BUCKET
    
    • --region: a supported Dataproc region where your Pub/Sub Lite topic and subscription reside.
    • --image-version: the cluster's image version , which determines the Apache Spark version installed on the cluster. Choose 1.5.x image release versions because the Pub/Sub Lite Spark Connector currently supports Apache Spark 2.4.x.
    • --scopes: enable API access to GCP services in the same project.
    • --enable-component-gateway: enable access to the Apache Spark web UI.
    • --bucket: a staging Cloud Storage bucket used to store cluster job dependencies, driver output, and cluster config files.
  5. Clone the quickstart repository and navigate to the sample code directory:

    git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git
    cd python-docs-samples/pubsublite/spark-streaming/
    

Writing to Pub/Sub Lite

The following example will:

  • create a rate source that generates consecutive numbers and timestamps formatted as spark.sql.Row
  • transform the data to match the required table schema by the Pub/Sub Lite Spark Connector's writeStream API
  • write the data to an existing Pub/Sub Lite topic
from pyspark.sql import SparkSession
from pyspark.sql.types import BinaryType, StringType
import uuid

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

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

# Create a RateStreamSource that generates consecutive numbers with timestamps:
# |-- timestamp: timestamp (nullable = true)
# |-- value: long (nullable = true)
sdf = spark.readStream.format("rate").option("rowsPerSecond", 1).load()

sdf = (
    sdf.withColumn("key", (sdf.value % 5).cast(StringType()).cast(BinaryType()))
    .withColumn("event_timestamp", sdf.timestamp)
    .withColumn("data", sdf.value.cast(StringType()).cast(BinaryType()))
    .drop("value", "timestamp")
)

sdf.printSchema()

query = (
    sdf.writeStream.format("pubsublite")
    .option(
        "pubsublite.topic",
        f"projects/{project_number}/locations/{location}/topics/{topic_id}",
    )
    # Required. Use a unique checkpoint location for each job.
    .option("checkpointLocation", "/tmp/app" + uuid.uuid4().hex)
    .outputMode("append")
    .trigger(processingTime="1 second")
    .start()
)

# Wait 60 seconds to terminate the query.
query.awaitTermination(60)
query.stop()

To submit the write job to Dataproc:

Console

  1. Upload the PySpark script to your Cloud Storage bucket.
    1. Go to the Cloud Storage console.
    2. Select your bucket.
    3. Use Upload files to upload the PySpark script that you intend to use.
  2. Submit the job to your Dataproc cluster:
    1. Go to the Dataproc console.
    2. Navigate to jobs.
    3. Click Submit job.
    4. Fill in the job details.
    5. Under Cluster, choose your cluster.
    6. Under Job, give a name to the job ID.
    7. For Job type, choose PySpark.
    8. For Main python file, provide the gsutil URI of the uploaded PySpark script that starts with gs://.
    9. For Jar files, use the Pub/Sub Lite Spark Connector's uber jar with dependencies in this link gs://spark-lib/pubsublite/pubsublite-spark-sql-streaming-LATEST-with-dependencies.jar . Alternatively, visit this Maven link , look for the jar with dependencies in the download options, and copy its link.
    10. For Arguments, if you use the full PySpark script from GitHub, enter --project_number=PROJECT_NUMBER, --location=PUBSUBLITE_LOCATION, --topic_id=TOPIC_ID ; if you copy the PySpark script above with the to-do's completed, leave it blank.
    11. Under Properties, enter key spark.master and value yarn.
    12. Click Submit.

gcloud

Use the gcloud dataproc jobs submit pyspark command to submit the job to Dataproc:

gcloud dataproc jobs submit pyspark spark_streaming_to_pubsublite_example.py \
    --region=$DATAPROC_REGION \
    --cluster=$CLUSTER_ID \
    --jars=gs://spark-lib/pubsublite/pubsublite-spark-sql-streaming-LATEST-with-dependencies.jar \
    --driver-log-levels=root=INFO \
    --properties=spark.master=yarn \
    -- --project_number=$PROJECT_NUMBER --location=$PUBSUBLITE_LOCATION --topic_id=$TOPIC
  • --region: the pre-selected Dataproc region.
  • --cluster: the Dataproc cluster name.
  • --jars: the Pub/Sub Lite Spark Connector's uber jar with dependencies in a public Cloud Storage bucket. You can also visit this link to download the uber jar with dependencies from Maven.
  • --driver-log-levels: set logging level to INFO at the root level.
  • --properties: use YARN resource manager for the Spark master.
  • --: provide the arguments required by the script.

If the writeStream operation succeeds, you should see log messages like the following locally as well as in the job details page in the Cloud Console:

INFO com.google.cloud.pubsublite.spark.PslStreamWriter: Committed 1 messages for epochId ..

Reading from Pub/Sub Lite

The following example will read messages from an existing Pub/Sub Lite subscription using the readStream API. The connector will output messages that conforms to the fixed table schema formatted as spark.sql.Row .

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()

To submit the read job to Dataproc:

Console

  1. Upload the PySpark script to your Cloud Storage bucket.
    1. Go to the Cloud Storage console.
    2. Select your bucket.
    3. Use Upload files to upload the PySpark script that you intend to use.
  2. Submit the job to your Dataproc cluster:
    1. Go to the Dataproc console.
    2. Navigate to jobs.
    3. Click Submit job.
    4. Fill in the job details.
    5. Under Cluster, choose your cluster.
    6. Under Job, give a name to the job ID.
    7. For Job type, choose PySpark.
    8. For Main python file, provide the gsutil URI of the uploaded PySpark script that starts with gs://.
    9. For Jar files, use the Pub/Sub Lite Spark Connector's uber jar with dependencies in this link gs://spark-lib/pubsublite/pubsublite-spark-sql-streaming-LATEST-with-dependencies.jar . Alternatively, visit this Maven link , look for the jar with dependencies in the download options, and copy its link.
    10. For Arguments, if you use the full PySpark script from GitHub, enter --project_number=PROJECT_NUMBER, --location=PUBSUBLITE_LOCATION, --subscription_id=SUBSCRIPTION_ID ; if you copy the PySpark script above with the to-do's completed, leave it blank.
    11. Under Properties, enter key spark.master and value yarn.
    12. Click Submit.

gcloud

Use the gcloud dataproc jobs submit pyspark command again to submit the job to Dataproc:

gcloud dataproc jobs submit pyspark spark_streaming_to_pubsublite_example.py \
    --region=$DATAPROC_REGION \
    --cluster=$CLUSTER_ID \
    --jars=gs://spark-lib/pubsublite/pubsublite-spark-sql-streaming-LATEST-with-dependencies.jar \
    --driver-log-levels=root=INFO \
    --properties=spark.master=yarn \
    -- --project_number=$PROJECT_NUMBER --location=$PUBSUBLITE_LOCATION --subscription_id=$SUBSCRIPTION
  • --region: the pre-selected Dataproc region.
  • --cluster: the Dataproc cluster name.
  • --jars: the Pub/Sub Lite Spark Connector's uber jar with dependencies in a public Cloud Storage bucket. You can also visit this link to download the uber jar with dependencies from Maven.
  • --driver-log-levels: set logging level to INFO at the root level.
  • --properties: use YARN resource manager for the Spark master.
  • --: provide required arguments for the script.

If the readStream operation succeeds, you should see log messages like the following locally as well as in the job details page in the Cloud Console:

+--------------------+---------+------+---+----+--------------------+--------------------+----------+
|        subscription|partition|offset|key|data|   publish_timestamp|     event_timestamp|attributes|
+--------------------+---------+------+---+----+--------------------+--------------------+----------+
|projects/50200928...|        0| 89523|  0|   .|2021-09-03 23:01:...|2021-09-03 22:56:...|        []|
|projects/50200928...|        0| 89524|  1|   .|2021-09-03 23:01:...|2021-09-03 22:56:...|        []|
|projects/50200928...|        0| 89525|  2|   .|2021-09-03 23:01:...|2021-09-03 22:56:...|        []|

Clean up

To avoid incurring charges to your Google Cloud account for the resources used in this page, follow these steps.

  1. Delete the topic and subscription.

    gcloud pubsub lite-topics delete $TOPIC
    gcloud pubsub lite-subscriptions delete $SUBSCRIPTION
    
  2. Delete the Dataproc cluster.

    gcloud dataproc clusters delete $CLUSTER_ID --region=$DATAPROC_REGION
    
  3. Remove the Cloud Storage bucket.

    gsutil rb gs://$BUCKET
    

What's next