没问题，这需求在Spark里完全可以实现，我给你分Spark SQL和Scala API两种常用方案，还会考虑到元素不存在的情况以及低版本兼容的处理：

1. 基于内置函数 array_position（Spark 2.4+ 推荐） #

array_position 函数会返回元素在数组中的1-based索引，我们只需要减1就能得到你要的0-based索引。如果元素不在数组里，array_position 会返回0，这时候可以用条件判断返回-1或者你需要的默认值：

SELECT
  text,
  featured_text,
  CASE
    WHEN array_position(featured_text, text) > 0 THEN array_position(featured_text, text) - 1
    ELSE -1 -- 元素不存在时的默认值，可根据需求修改
  END AS index
FROM your_table_name;

针对你给出的示例，执行后会得到：

2. 低版本Spark兼容方案（Spark < 2.4） #

如果你的Spark版本低于2.4，没有array_position函数，可以通过posexplode来实现：

SELECT
  t.text,
  t.featured_text,
  COALESCE(MAX(p.pos), -1) AS index
FROM your_table_name t
LEFT JOIN LATERAL VIEW posexplode(t.featured_text) p AS pos, val
ON p.val = t.text
GROUP BY t.text, t.featured_text;

这个方法通过posexplode把数组拆成索引和元素的行，匹配text列的值后聚合取对应索引，元素不存在时返回-1。

优先使用内置函数（性能比自定义UDF更好）：

import org.apache.spark.sql.functions._

// 假设你的DataFrame名为df
val resultDF = df.withColumn(
  "index",
  when(
    array_position(col("featured_text"), col("text")) > 0,
    array_position(col("featured_text"), col("text")) - 1
  ).otherwise(-1) // 元素不存在时的默认值
)

// 查看结果
resultDF.show()

如果是低版本Spark，也可以用DataFrame API结合posexplode处理：

val resultDF = df
  .select(
    col("text"),
    col("featured_text"),
    posexplode(col("featured_text")).alias("pos", "val")
  )
  .filter(col("val") === col("text"))
  .select(col("text"), col("featured_text"), col("pos").alias("index"))
  .rightJoin(df, Seq("text", "featured_text"))
  .na.fill(-1, Seq("index"))

这样处理后，同样能得到你需要的索引结果，元素不存在时会自动填充-1。

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