Spark结构化流如何优雅忽略Avro格式的Kafka主题文件夹

问题背景 #

我有一个通用Spark结构化流作业，逻辑是监控顶级文件夹下的Kafka主题子文件夹，将每个主题的数据写入独立的Delta输出文件夹，同时完成数据压缩与外部表创建，目标是实现作业通用性，无需为每个Kafka主题单独编写流处理任务。

由于历史原因，部分Kafka主题文件夹存储Avro格式文件，部分为Parquet格式，当前仅需处理Parquet格式主题，但需要优雅忽略Avro格式的主题文件夹。

当前困境 #

• 使用全局异常处理会掩盖真实错误（比如Parquet文件本身的损坏、权限问题等），无法区分是格式不兼容还是其他严重问题
• 尝试过指定format和pathGlobFilter，但Avro文件夹过滤后为空会触发异常，且主题文件夹无命名规则可提前区分格式（每个主题仅含一种格式）

现有代码片段 #

top_level_folder_path= f"abfss://{sourcecontainer}@{datalakename}.dfs.core.windows.net/toplevelfolder"

for datahub_domain in dbutils.fs.ls(top_level_folder_path):
    for datahub_topic in topicsearchpath:
        # 推导变量逻辑省略
        ...............................
        # CloudFiles配置
        cloudfile = {
                    "cloudFiles.format": "parquet",
                    "cloudFiles.includeExistingFiles": "true",
                    "cloudFiles.inferColumnTypes": "true",
                    "cloudFiles.schemaLocation": f"abfss://raw@{datalakename}.dfs.core.windows.net/{originaldomainname}/autoloader/schemas/{originaltopicname}/",
                    "cloudFiles.schemaEvolutionMode": "addNewColumns",
                    "cloudFiles.allowOverwrites": "true",
                    "ignoreCorruptFiles": "true",
                    "ignoreMissingFiles": "true",
                }
        try:
            df = (
                spark.readStream.format("cloudFiles")
                .options(**cloudfile)
                .load(datahub_topic.path)
            )
            dstreamQuery = (
                df.writeStream.format("delta")
                .outputMode("append")
                .queryName(f"{schema_name}_raw_{table_name}")
                .option(
                    "checkpointLocation",
                    f"abfss://raw@{datalakename}.dfs.core.windows.net/autoloader/checkpoint/{originaldomainname}/{originaltopicname}/",
                )
                .option("mergeSchema", "true")
                .partitionBy("Year", "Month", "Day")
                .trigger(availableNow=True)
                .start(
                    f"abfss://raw@{datalakename}.dfs.core.windows.net/{originaldomainname}/delta/{originaltopicname}"
                )
            )
            while len(spark.streams.active) > 0:
                spark.streams.awaitAnyTermination()
        except Exception as e: # 不想用这种全局捕获
            logger.warning(f"Error reading stream: {str(e)}") # 会掩盖真实错误

需求 #

作业为每个主题文件夹启动独立流任务，需解决如何在读取Parquet流时精准跳过Avro主题文件夹，避免全局异常处理掩盖真实错误。

解决方案 #

1. 前置检测：先验证主题文件夹内的文件格式 #

在启动流任务前，先遍历主题文件夹下的文件，通过扩展名判断格式，仅处理存在.parquet文件的文件夹：

def is_parquet_topic(topic_path):
    # 列出文件夹下的文件（递归或仅一级，根据实际存储结构调整）
    files = dbutils.fs.ls(topic_path)
    # 检查是否存在parquet文件
    has_parquet = any(file.name.endswith(".parquet") for file in files if not file.isDir())
    # 每个主题仅一种格式，可同时确认无avro文件
    has_avro = any(file.name.endswith(".avro") for file in files if not file.isDir())
    return has_parquet and not has_avro

# 在循环中加入检测逻辑
for datahub_domain in dbutils.fs.ls(top_level_folder_path):
    for datahub_topic in topicsearchpath:
        if not is_parquet_topic(datahub_topic.path):
            logger.info(f"Skipping non-parquet topic folder: {datahub_topic.path}")
            continue
        # 后续流处理逻辑不变
        cloudfile = { ... }
        try:
            df = spark.readStream.format("cloudFiles").options(**cloudfile).load(datahub_topic.path)
            # ... Delta写入逻辑 ...
        except Exception as e:
            # 此时抛出的异常均为Parquet处理相关的真实错误，直接抛出或针对性处理
            raise e

2. 结合pathGlobFilter+空目录检测 #

用pathGlobFilter过滤.parquet文件，同时在加载前检查过滤后的文件数量，避免空目录触发异常：

cloudfile = {
    "cloudFiles.format": "parquet",
    "cloudFiles.pathGlobFilter": "*.parquet",  # 仅匹配parquet文件
    # 其他配置保持不变...
}

# 先检查主题文件夹下是否有可加载的parquet文件
try:
    parquet_count = spark.read.format("parquet").option("pathGlobFilter", "*.parquet").load(datahub_topic.path).count()
    if parquet_count == 0:
        logger.info(f"No parquet files found in {datahub_topic.path}, skipping")
        continue
except Exception as e:
    logger.warning(f"Failed to check parquet files in {datahub_topic.path}: {str(e)}")
    continue

# 确认有parquet文件后再启动流任务
df = spark.readStream.format("cloudFiles").options(**cloudfile).load(datahub_topic.path)
# ... Delta写入逻辑 ...

3. 捕获特定异常而非全局异常 #

如果前置检测无法覆盖所有场景，可针对性捕获格式不兼容的异常，其他异常正常抛出：

from pyspark.sql.utils import AnalysisException

try:
    df = spark.readStream.format("cloudFiles").options(**cloudfile).load(datahub_topic.path)
    # ... Delta写入逻辑 ...
except AnalysisException as e:
    error_msg = str(e).lower()
    # 匹配Parquet格式不兼容的特征错误信息
    if "parquet" in error_msg and ("not a parquet file" in error_msg or "invalid parquet" in error_msg):
        logger.warning(f"Skipping Avro topic folder: {datahub_topic.path}, error: {e}")
    else:
        # 其他分析异常（如schema问题、权限错误）直接抛出
        raise e
except Exception as e:
    # 非分析异常（如网络、IO错误）直接抛出
    raise e

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