如何用Spark/PySpark获取每位客户的首航班出发地与末航班目的地

方法一：基于行程节点的出入度判断 #

因为每个客户的行程是连贯路径（比如示例中的texas→newyork→delhi→japan），首站origin不会出现在该客户所有航班的destination列表中，末站destination不会出现在该客户所有航班的origin列表中。可以通过分组聚合+过滤实现：

from pyspark.sql import SparkSession
from pyspark.sql.functions import collect_set, array_contains

# 初始化SparkSession
spark = SparkSession.builder.appName("FlightPath").getOrCreate()

# 模拟输入DataFrame
data = [(1, "f12", "texas", "newyork"),
        (1, "f21", "newyork", "delhi"),
        (1, "f08", "delhi", "japan")]
df = spark.createDataFrame(data, ["cid", "fid", "origin", "destination"])

# 1. 按cid分组，收集该客户所有的origin和destination集合
agg_df = df.groupBy("cid")\
    .agg(collect_set("origin").alias("origin_set"),
         collect_set("destination").alias("dest_set"))

# 2. 筛选首站（origin不在dest_set里）和末站（destination不在origin_set里）
first_origin = df.join(agg_df, on="cid")\
    .filter(~array_contains(agg_df.dest_set, df.origin))\
    .select("cid", "origin").withColumnRenamed("origin", "first_origin")

last_dest = df.join(agg_df, on="cid")\
    .filter(~array_contains(agg_df.origin_set, df.destination))\
    .select("cid", "destination").withColumnRenamed("destination", "last_destination")

# 3. 合并结果集
result = first_origin.join(last_dest, on="cid")
result.show()

执行后输出：

+---+------------+---------------+
|cid|first_origin|last_destination|
+---+------------+---------------+
|  1|       texas|           japan|
+---+------------+---------------+

方法二：用窗口函数（若有航班时间字段） #

如果数据包含航班出发时间（比如departure_time），可以直接按cid分组、时间排序，取第一条的origin和最后一条的destination，逻辑更直观：

from pyspark.sql.window import Window
from pyspark.sql.functions import first, last

# 模拟带时间字段的输入数据
data_with_time = [(1, "f12", "texas", "newyork", "2024-01-01 08:00"),
                  (1, "f21", "newyork", "delhi", "2024-01-02 10:00"),
                  (1, "f08", "delhi", "japan", "2024-01-03 12:00")]
df_with_time = spark.createDataFrame(data_with_time, ["cid", "fid", "origin", "destination", "departure_time"])

# 定义窗口：按cid分组，按出发时间升序排列
window_spec = Window.partitionBy("cid").orderBy("departure_time")

# 聚合提取首末站点
result = df_with_time.withColumn("first_origin", first("origin").over(window_spec))\
    .withColumn("last_destination", last("destination").over(window_spec))\
    .groupBy("cid")\
    .agg(first("first_origin").alias("origin"),
         last("last_destination").alias("destination"))

result.show()

关于你尝试的Join操作 #

上面的方法一已经用到Join，但核心是先通过聚合得到每个客户的站点集合，再用集合判断筛选首末站，最后合并结果。如果直接用原表自关联（比如匹配origin等于另一条的destination），会把中间节点也拉出来，效率和准确性都不如先聚合再过滤的方式。

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