Spark 3.4.3 Kafka Reader正则匹配Topic异常求助

问题分析 #

你遇到的核心问题是Spark Kafka数据源的subscribePattern参数使用Java正则引擎，和regex101默认的PCRE引擎存在行为差异，同时你的正则写法也存在匹配逻辑问题：

1. 第一个正则k8s-v2.in.mongo.mongo-conn.ninja.xyz.PolicyDetail中的.是正则通配符，会匹配任意单个字符，导致所有包含该子串结构的Topic都被匹配（范围覆盖绝大多数Topic）。
2. 第二个正则^(k8s-v2.in.mongo.mongo-conn.ninja.xyz.PolicyDetail)$在PCRE引擎中能精准匹配单个Topic，但在Spark中出现异常匹配两个Topic，本质是未正确转义.，且正则逻辑未覆盖第二个Topic的后缀。

正确解决方案 #

方案1：适配Java正则的精准匹配规则 #

要同时匹配指定的两个Topic，使用符合Java正则语法的规则：

^k8s-v2\.in\.mongo\.mongo-conn\.ninja\.xyz\.PolicyDetail(History)?$

• 转义所有.：Java正则中.是通配符，必须用\.表示实际的点字符。
• (History)?表示可选的History后缀，同时匹配PolicyDetail和PolicyDetailHistory两个Topic。
• ^和$锚点确保完全匹配Topic名称，避免部分匹配其他无关Topic。

修改后的Spark代码：

spark.readStream.format("kafka")
            .option("kafka.bootstrap.servers", "host:9092")
            .option("subscribePattern", "^k8s-v2\\.in\\.mongo\\.mongo-conn\\.ninja\\.xyz\\.PolicyDetail(History)?$")
            .option("startingOffsets", "earliest")
            .option("failOnDataLoss", "false")
            .option("maxOffsetsPerTrigger", 1000000)
            .option("group.id", "group1")
            .load()

注：Scala/Java字符串中\需双重转义为\\，因此正则中的\.要写成\\.。

方案2：直接订阅多个Topic（更简单可靠） #

如果目标Topic数量固定，直接用subscribe参数指定多个Topic，比正则更直观且不易出错：

spark.readStream.format("kafka")
            .option("kafka.bootstrap.servers", "host:9092")
            .option("subscribe", "k8s-v2.in.mongo.mongo-conn.ninja.xyz.PolicyDetail,k8s-v2.in.mongo.mongo-conn.ninja.xyz.PolicyDetailHistory")
            .option("startingOffsets", "earliest")
            .option("failOnDataLoss", "false")
            .option("maxOffsetsPerTrigger", 1000000)
            .option("group.id", "group1")
            .load()

关键注意事项 #

• Spark Kafka的subscribePattern依赖Java正则引擎，编写正则时需遵循Java正则语法，而非PCRE。
• 字符串中的特殊字符（如.、-）必须正确转义，避免意外匹配。
• 当Topic数量较少时，优先使用subscribe直接指定，性能和可读性更优。

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