我正在消费两个币安的数据流：一个交易流和一个K 线（蜡烛图）流。这是我在Spark作业中使用的模式/结构：

========================================================================
# TRADE STREAM SCHEMA (MICRO VIEW)
# ========================================================================
trade_schema = StructType([
    StructField("e", StringType(), True),
    StructField("s", StringType(), True),
    StructField("t", LongType(), True),
    StructField("p", StringType(), True),
    StructField("q", StringType(), True),
    StructField("T", LongType(), True),
    StructField("m", BooleanType(), True)
])

parsed_trade_df = (
    trade_raw_df
    .select(
        from_json(
            col("value").cast("string"),
            trade_schema
        ).alias("json")
    )
    .filter(col("json").isNotNull())
    .select(
        col("json.e").alias("event_type"),
        col("json.s").alias("symbol"),
        col("json.t").alias("trade_id"),
        col("json.p").cast(DecimalType(18, 2)).alias("price"),
        col("json.q").cast(DecimalType(18, 6)).alias("quantity"),
        col("json.T").alias("trade_time_ms"),
        col("json.m").alias("is_buyer_maker")
    )
    .filter(col("event_type") == "trade")
    .filter(col("trade_time_ms").isNotNull())
    .filter(col("price").isNotNull())
    .filter(col("quantity").isNotNull())
)
parsed_trade_df.printSchema()
# ========================================================================
# TRADE TRANSFORMATION
# ========================================================================
trade_df = (
    parsed_trade_df
    .withColumn(
        "event_time",
        (col("trade_time_ms") / 1000).cast("timestamp")
    )
    .withColumn(
        "total_value_usd",
        col("price") * col("quantity")
    )
)

当Spark容器启动时，我得到以下错误：

26/06/23 18:28:01 INFO StandaloneSchedulerBackend: SchedulerBackend is ready for scheduling beginning after reached minRegisteredResourcesRatio: 0.0

Traceback (most recent call last):

  File "/opt/spark/app/spark_streaming.py", line 95, in <module>

    trade_raw_df

  File "/opt/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 3223, in select

  File "/opt/spark/python/lib/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__

  File "/opt/spark/python/lib/pyspark.zip/pyspark/errors/exceptions/captured.py", line 185, in deco

pyspark.errors.exceptions.captured.AnalysisException: [AMBIGUOUS_REFERENCE_TO_FIELDS] Ambiguous reference to the field `t`. It appears 2 times in the schema.

[Tue 23 Jun 2026 06:28:03 PM UTC] Spark exited with code 1

Retrying in 10 seconds...

回溯指向我JSON解析步骤中的一个 .select(...) 操作。

让我困惑的是，交易模式只有一个字段，名为 t，而K 线模式在 k 结构中嵌套了另一个 t。我以为Spark应该能够区分 json.t 与 json.k.t。

在Spark中通常是什么原因会导致这个错误？

示例交易：

{
  "e": "trade",
  "E": 1782371994016,
  "s": "BTCUSDT",
  "t": 6442492017,
  "p": "61849.00000000",
  "q": "0.00016000",
  "T": 1782371994015,
  "m": false
}

kline数据流

{
  "e": "kline",
  "E": 1782371994016,
  "s": "BTCUSDT",
  "k": {
    "t": 1782371940000,
    "T": 1782371999999,
    "i": "1m",
    "o": "61802.00000000",
    "c": "61848.99000000",
    "h": "61849.00000000",
    "l": "61799.00000000",
    "v": "7.06707000",
    "n": 1618,
    "x": false,
    "q": "436860.47143120"
  }
}

解决方案

这是一个简单的大小写敏感问题。默认情况下 spark.sql.caseSensitive 被设置为 false

>>> 
>>> from pyspark.sql.types import StructType, StructField, StringType, LongType, BooleanType
>>> from pyspark.sql.functions import from_json, col
>>> from pyspark.sql import Row
>>> 
>>> json_str = '{"e":"trade","E":1782371994016,"s":"BTCUSDT","t":6442492017,"p":"61849.00","q":"0.00016","T":1782371994015,"m":false}'
>>> df = spark.createDataFrame([Row(value=json_str)])
>>> 
>>> schema = StructType([
...     StructField("e", StringType()),
...     StructField("s", StringType()),
...     StructField("t", LongType()),
...     StructField("p", StringType()),
...     StructField("q", StringType()),
...     StructField("T", LongType()),
...     StructField("m", BooleanType())
... ])
>>> 
>>> spark.conf.set('spark.sql.caseSensitive', 'false')
>>> #       this is the default value          ^^^^^
>>> 
>>> df.select(from_json(col("value"), schema).alias("json")).select("json.t", "json.T").show()
....
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "workspaces/dlh-databricks/fxc/.venv/lib64/python3.12/site-packages/pyspark/sql/classic/dataframe.py", line 991, in select
    jdf = self._jdf.select(self._jcols(*cols))
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
....
    raise converted from None
pyspark.errors.exceptions.captured.AnalysisException: [AMBIGUOUS_REFERENCE_TO_FIELDS] Ambiguous reference to the field `t`. It appears 2 times in the schema. SQLSTATE: 42000
>>>

修改配置：

>>> 
>>> spark.conf.set('spark.sql.caseSensitive', 'true')
>>> df.select(from_json(col("value"), schema).alias("json")).select("json.t", "json.T").show()
+----------+-------------+                                                      
|         t|            T|
+----------+-------------+
|6442492017|1782371994015|
+----------+-------------+

>>>

就我而言，你应该只使用不同的列名来替代配置，而不是修改配置。不过，随你便。