Spark Streaming读取Pub/Sub Lite消息无法实时接收，被批量分组

在GCP的Spark集群上尝试实时接收Pub/Sub Lite的消息，但消息被按每分钟分组批量返回。

代码实现 #

producer.py #

import random
import time
from proj_BOLSA import settings
from google.cloud.pubsublite.cloudpubsub import PublisherClient
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    MessageMetadata,
    TopicPath,
)
import os
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/home/vm-sergiolr-development/Desktop/projecteemo_code/proj_BOLSA/credentials/gcp_authentication.json"

regional = True

if regional:
    location = CloudRegion(settings.REGION)
else:
    location = CloudZone(CloudRegion(settings.REGION), settings.ZONE)

topic_path = TopicPath(settings.PROJECT_NUMBER, location, settings.TOPIC)

# PublisherClient() must be used in a `with` block or have __enter__() called before use.
with PublisherClient() as publisher_client:
    for i in range(6000):
        data = "number: "+str(random.randint(0, 300))
        api_future = publisher_client.publish(topic_path, data.encode("utf-8"))
        # result() blocks. To resolve API futures asynchronously, use add_done_callback().
        message_id = api_future.result()
        message_metadata = MessageMetadata.decode(message_id)
        print(
            f"Published {data} to {topic_path} with partition {message_metadata.partition.value} and offset {message_metadata.cursor.offset}."
        )
        time.sleep(20)

consumer.py #

from pyspark.sql import SparkSession
from pyspark.sql import functions as f
from pyspark.sql.types import StringType
from pyspark.sql.types import StructType, StructField, StringType, ArrayType
from pyspark.sql.functions import from_json, col
# TODO(developer):
project_number = xxxxxxxx
location = "europe-west1"
subscription_id = "s_producer"


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}",
    )
    .option("rowsPerSecond", 1).load()
)


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

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

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

运行结果 #

-------------------------------------------
Batch: 1
-------------------------------------------
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+
|subscription                                                         |partition|offset|key|data       |publish_timestamp         |event_timestamp|attributes|
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5942  |[] |number: 74 |2022-08-05 08:42:47.796738|null           |{}        |
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5943  |[] |number: 288|2022-08-05 08:43:07.849063|null           |{}        |
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5944  |[] |number: 156|2022-08-05 08:43:27.952513|null           |{}        |
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+

-------------------------------------------
Batch: 2
-------------------------------------------
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+
|subscription                                                         |partition|offset|key|data       |publish_timestamp         |event_timestamp|attributes|
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5945  |[] |number: 162|2022-08-05 08:43:48.00867 |null           |{}        |
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5946  |[] |number: 262|2022-08-05 08:44:08.062032|null           |{}        |
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5947  |[] |number: 59 |2022-08-05 08:44:28.11492 |null           |{}        |
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+

-------------------------------------------
Batch: 3
-------------------------------------------
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+
|subscription                                                         |partition|offset|key|data       |publish_timestamp         |event_timestamp|attributes|
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5948  |[] |number: 54 |2022-08-05 08:44:48.168997|null           |{}        |
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5949  |[] |number: 206|2022-08-05 08:45:08.225344|null           |{}        |
|projects/658599344059/locations/europe-west1/subscriptions/s_producer|0        |5950  |[] |number: 109|2022-08-05 08:45:28.328074|null           |{}        |
+---------------------------------------------------------------------+---------+------+---+-----------+--------------------------+---------------+----------+

问题分析与解决方案 #

核心问题 #

1. Pub/Sub Lite Spark连接器默认批量策略：连接器默认会攒够一定数量或字节数的消息才返回，目的是优化吞吐量，但低消息量场景下会导致延迟。
2. rowsPerSecond参数误用：该参数是限制Spark每秒处理的行数，而非控制消息拉取频率，反而会造成消息限流积压。
3. Trigger设置未配合拉取超时：虽然设置了1秒的处理触发间隔，但连接器如果未拉取到足够消息会一直等待，直到达到默认批量阈值。

修正后的Consumer配置 #

调整连接器拉取参数，强制尽快返回新消息：

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

# TODO(developer):
project_number = xxxxxxxx
location = "europe-west1"
subscription_id = "s_producer"

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}",
    )
    # 每个微批最多拉取1条消息
    .option("pubsublite.maxBatchSize", 1)
    # 每个微批最大字节数设为1KB，确保小消息能及时返回
    .option("pubsublite.maxBatchBytes", 1024)
    # 拉取超时1秒，超时后即使没有足够消息也返回
    .option("pubsublite.pollTimeoutMs", 1000)
    .load()
)

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

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

query.awaitTermination(120)
query.stop()

额外说明 #

• 你的Producer每20秒发送一条消息，修改后大部分1秒触发的微批会是空的，但新消息到达时，下一个触发周期会立即拉取并输出，实现近似实时的效果。
• 后续消息量增大时，可适当调大maxBatchSize和maxBatchBytes，平衡延迟与吞吐量。

内容的提问来源于stack exchange，提问作者user9548476