如何以编程方式排查Synapse无服务器SQL池含坏数据的视图

直接从数据源层面排查，能精准定位到含0001-01-01日期/时间戳的异常文件，效率比依赖Synapse查询更高。

1. 读取Parquet/Delta文件 #

针对Parquet格式：

# 替换为你的ADLS路径，支持通配符批量读取
parquet_df = spark.read.parquet("abfss://<容器名>@<存储账户>.dfs.core.windows.net/<Parquet文件路径>")

针对Delta格式：

delta_df = spark.read.format("delta").load("abfss://<容器名>@<存储账户>.dfs.core.windows.net/<Delta文件路径>")

2. 扫描日期/时间字段，定位异常文件 #

遍历所有日期/时间类型字段，筛选出含0001-01-01的记录，并关联到对应的源文件：

from pyspark.sql.functions import col, input_file_name

# 自动识别所有日期、时间戳类型字段
datetime_cols = [
    field.name for field in parquet_df.schema.fields 
    if str(field.dataType).startswith(('TimestampType', 'DateType'))
]

# 添加源文件路径列，用于定位异常文件
df_with_file = parquet_df.withColumn("source_file", input_file_name())

# 逐个字段检查坏数据
for col_name in datetime_cols:
    bad_records = df_with_file.filter(col(col_name) == "0001-01-01")
    if bad_records.count() > 0:
        print(f"字段 {col_name} 存在坏数据，涉及文件：")
        # 去重输出异常文件列表
        for file in bad_records.select("source_file").distinct().collect():
            print(file.source_file)

3. 批量处理视图对应的文件路径 #

从Synapse导出视图关联的ADLS路径，再批量传入Databricks扫描：
先在Synapse无服务器SQL池执行以下SQL，导出视图与对应ADLS路径的映射：

SELECT 
    v.name AS view_name,
    -- 解析OPENROWSET中的ADLS路径，需根据你的视图定义调整正则逻辑
    SUBSTRING(m.definition, 
              CHARINDEX('''abfss://', m.definition) + 1, 
              CHARINDEX('''', m.definition, CHARINDEX('''abfss://', m.definition)+1) - CHARINDEX('''abfss://', m.definition) -1) AS adls_path
FROM sys.views v
JOIN sys.sql_modules m ON v.object_id = m.object_id
WHERE v.name IN ('<你的视图列表>')

将导出的路径列表导入Databricks，循环执行上述扫描逻辑即可。

如果需要直接确认视图是否因该Bug无法查询，可通过Databricks连接Synapse批量执行查询，捕获特定异常：

1. 配置Synapse JDBC连接 #

jdbc_url = "jdbc:sqlserver://<Synapse工作区>.sql.azuresynapse.net:1433;database=<数据库名>;encrypt=true;trustServerCertificate=false;hostNameInCertificate=*.sql.azuresynapse.net;loginTimeout=30;"
connection_properties = {
    "user": "<用户名>",
    "password": "<密码>",
    "driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver"
}

2. 批量查询视图，捕获异常 #

view_list = ["view1", "view2", "view3"]  # 替换为你的视图列表
problem_views = []

for view in view_list:
    try:
        # 执行count查询触发报错
        spark.read.jdbc(url=jdbc_url, table=f"(SELECT COUNT(*) FROM {view}) AS tmp", properties=connection_properties).count()
    except Exception as e:
        # 根据Synapse报错关键词判断是否为目标Bug
        if "0001-01-01" in str(e) and "timestamp" in str(e):
            problem_views.append(view)
            print(f"视图 {view} 存在坏数据问题")

print("存在问题的视图列表：", problem_views)

注意事项 #

• Delta Lake文件可通过input_file_name()获取底层Parquet文件路径，同样适用扫描逻辑；
• 大规模文件扫描可开启分区并行处理，提升效率；
• 方案一适合定位根源文件，方案二适合直接验证视图可用性，可根据需求选择。

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