同一查询在Databricks与Snowflake中返回结果不一致问题排查

以下是可能导致结果不一致的关键点及对应修复建议：

1. 冗余的DISTINCT与GROUP BY组合 #

你的子查询同时使用了GROUP BY和DISTINCT，属于冗余写法，但不同SQL引擎的处理逻辑存在差异：

• GROUP BY B.VISITOR_ID, CREATE_TS, B.VISIT_ID已经保证分组后这三个字段的唯一性，DISTINCT完全多余。
• Snowflake可能直接忽略DISTINCT，而Spark会在分组后额外执行去重操作，过滤掉部分分组数据，最终导致SUM结果偏小。

修复建议：删除子查询中的DISTINCT关键字，修改后的子查询如下：

SELECT B.VISITOR_ID, TO_TIMESTAMP_NTZ(CREATE_TS) as CREATE_TS, B.VISIT_ID, COUNT(A.ORDER_ID_V32) as ORDER_ID
FROM EDM_VIEWS_QA.DW_VIEWS.CLICK_STREAM_SHOP A
JOIN EDM_VIEWS_QA.DW_VIEWS.CLICK_STREAM_OTHER B
ON A.CLICK_STREAM_INTEGRATION_ID = B.CLICK_STREAM_INTEGRATION_ID
JOIN EDM_VIEWS_QA.DW_VIEWS.CLICK_STREAM_METRICS C
ON A.CLICK_STREAM_INTEGRATION_ID = C.CLICK_STREAM_INTEGRATION_ID
WHERE DATE(B.CREATE_TS) = '2023-09-08'
AND B.DW_CURRENT_VERSION_IND='TRUE' AND C.DW_CURRENT_VERSION_IND='TRUE'
GROUP BY B.VISITOR_ID, CREATE_TS, B.VISIT_ID

2. TO_TIMESTAMP_NTZ函数的跨引擎兼容性 #

Snowflake和Spark的TO_TIMESTAMP_NTZ函数在解析精度、时区处理上存在细微差异：

• 若原始CREATE_TS为字符串类型，两者对格式的容错性不同；若为带时区的时间戳，转换为无时区类型时的截断逻辑不一致，会导致分组键不匹配，最终计数结果偏差。

修复建议：

• 若CREATE_TS本身已是时间戳类型，直接使用B.CREATE_TS作为分组键，避免额外转换；
• 若必须转换，统一使用显式类型转换语法：CAST(B.CREATE_TS AS TIMESTAMP_NTZ)，确保两边逻辑完全一致。

3. DATE()函数的时区差异 #

DATE(B.CREATE_TS)的结果依赖会话时区设置：

• 直接在Snowflake客户端执行时的会话时区，可能与Databricks中Snowflake连接器使用的时区不同，导致过滤的日期范围不一致（比如带时区的CREATE_TS在不同时区下转换为日期可能跨天）。

修复建议：改用无歧义的日期范围过滤，避免依赖时区：

WHERE B.CREATE_TS >= '2023-09-08 00:00:00' AND B.CREATE_TS < '2023-09-09 00:00:00'

同时在Databricks的Snowflake连接器中显式设置与Snowflake客户端一致的会话时区：

BIMDF2 = spark.read.format("snowflake")
.options(**db_options)
.option("session_parameters", {"TIMEZONE": "UTC"})  # 替换为你的Snowflake会话时区
.option("query", TEST_Q)
.load()

4. 连接后的重复数据计数问题 #

三个表通过CLICK_STREAM_INTEGRATION_ID连接时，可能存在一对多关联，导致同一条ORDER_ID_V32被多次统计：

• 不同引擎对分组计数的处理逻辑不同，Snowflake可能自动忽略重复的ORDER_ID_V32，而Spark会重复计数。

修复建议：将COUNT(A.ORDER_ID_V32)改为COUNT(DISTINCT A.ORDER_ID_V32)，确保每个ORDER_ID_V32只被计数一次；同时可以单独查询子查询的结果行数，对比Snowflake和Databricks返回的行数是否一致，定位是子查询结果差异还是SUM计算的问题。

5. Snowflake连接器的会话参数差异 #

Databricks中Snowflake连接器默认的会话参数（比如TIMESTAMP_TYPE_MAPPING、QUOTED_IDENTIFIERS_IGNORE_CASE等），可能与你直接在Snowflake客户端使用的参数不同，导致查询逻辑执行差异。

修复建议：在连接器中显式设置与Snowflake客户端一致的会话参数，例如：

session_params = {
    "TIMEZONE": "UTC",
    "TIMESTAMP_TYPE_MAPPING": "TIMESTAMP_NTZ",
    "QUOTED_IDENTIFIERS_IGNORE_CASE": "TRUE"
}
BIMDF2 = spark.read.format("snowflake")
.options(**db_options)
.option("session_parameters", session_params)
.option("query", TEST_Q)
.load()

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