一些缺失部分：
不能用spark执行impala函数。
有一个具有相同名称和语法的配置单元udf可以与spark一起使用，但它没有本机实现和函数 Package 器。这就是为什么可以使用sql调用它 HiveContext / SparkSession 有Hive的支持。
一般来说，它应该工作得很好：

spark.sql("""SELECT parse_url(
    'http://example.com?extensionId=foo', 'QUERY', 'extensionId'
)""").show()

+-----------------------------------------------------------------+
|parse_url(http://example.com?extensionId=foo, QUERY, extensionId)|
+-----------------------------------------------------------------+
|                                                              foo|
+-----------------------------------------------------------------+

以及 NULL 输出意味着给定部分无法匹配：

spark.sql("""SELECT parse_url(
    'http://example.com?bar=foo', 'QUERY', 'extensionId'
)""").show()

+---------------------------------------------------------+
|parse_url(http://example.com?bar=foo, QUERY, extensionId)|
+---------------------------------------------------------+
|                                                     null|
+---------------------------------------------------------+

使用自定义项可以获得类似的结果，但速度会明显减慢。

from typing import Dict
from urllib.parse import parse_qsl, urlsplit
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, MapType

def parse_args(col: str) -> Dict[str, str]:
    """
    http://stackoverflow.com/a/21584580/6910411
    """
    try:
        return dict(parse_qsl(urlsplit(col).query))
    except:
        pass

parse_args_ = udf(parse_args, MapType(StringType(), StringType()))

数据定义为：

df = sc.parallelize([
    ("http://example.com?bar=foo", ),
    ("http://example.com?extensionId=foo", ),
]).toDF(["url"])

其用途如下：

df.select(parse_args_("url")["extensionId"]).show()

结果是：

+----------------------------+
|parse_args(url)[extensionId]|
+----------------------------+
|                        null|
|                         foo|
+----------------------------+