PySpark：面试题（编码题）— 第一部分

参加一次 PySpark 面试可能是一项具有挑战性的经历，尤其是在面对基于场景的题目时，这些问题不仅测试一个人的理论知识，还测试其实际问题解决能力。

在这篇文章中，我将分享我在面试中遇到的一些个人问题以及我认为重要的问题。我相信这些问题将有助于了解这些类型的问题，在线评估或面对面面试中可能遇到的问题类型。

给定用户活动数据的点击流，找到每个点击事件的相关用户会话。

click_time | user_id
2018-01-01 11:00:00 | u1
2018-01-01 12:00:00 | u1
2018-01-01 13:00:00 | u1
2018-01-01 13:00:00 | u1
2018-01-01 14:00:00 | u1
2018-01-01 15:00:00 | u1
2018-01-01 11:00:00 | u2
2018-01-02 11:00:00 | u2

会话定义：
1. 会话在30分钟不活跃后过期，不再生成点击流
2. 会话保持活跃总共2小时的时间限制

（注：此处“Sol”保留原文，具体含义需根据上下文确定。）

    from pyspark.sql import SparkSession  
    from pyspark.sql.functions import col, unix_timestamp, lag, when, lit, concat, sum, monotonically_increasing_id  
    from pyspark.sql.window import Window  

    # 创建一个SparkSession  
    spark = SparkSession.builder \  
        .appName("ClickStreamSession") \  
        .getOrCreate()  

    # 定义点击流数据的模式  
    schema = "click_time STRING, user_id STRING"  

    # 点击流数据样本数据  
    data = [  
        ("2018-01-01 11:00:00", "u1"),  
        ("2018-01-01 12:00:00", "u1"),  
        ("2018-01-01 13:00:00", "u1"),  
        ("2018-01-01 13:00:00", "u1"),  
        ("2018-01-01 14:00:00", "u1"),  
        ("2018-01-01 15:00:00", "u1"),  
        ("2018-01-01 11:00:00", "u2"),  
        ("2018-01-02 11:00:00", "u2")  
    ]  

    # 从给定的样本数据创建一个DataFrame  
    clickstream_df = spark.createDataFrame(data, schema=schema)  

    # 将click_time转换为Unix时间戳，以便进行更简单的计算和处理  
    clickstream_df = clickstream_df.withColumn("click_timestamp", unix_timestamp("click_time"))  

    session_window = Window.partitionBy("user_id").orderBy("click_timestamp")  

    # 使用lag函数获取前一行的点击时间戳  
    clickstream_df = clickstream_df.withColumn("prev_click_timestamp", lag("click_timestamp", 1).over(session_window))  

    # 计算点击时间之间的差异并将该差异除以60  
    clickstream_df = clickstream_df.withColumn("timestamp_diff", (col("click_timestamp")-col("prev_click_timestamp"))/60)  

    # 将null值更新为0  
    clickstream_df = clickstream_df.withColumn("timestamp_diff", when(col("timestamp_diff").isNull(), 0).otherwise(col("timestamp_diff")))  

    # 检查是否为新会话  
    clickstream_df = clickstream_df.withColumn("session_new", when(col("timestamp_diff") > 30, 1).otherwise(0))  

    # 新会话标识符  
    clickstream_df = clickstream_df.withColumn("session_new_name", concat(col("user_id"), lit("--S"), sum(col("session_new")).over(session_window)))  
    clickstream_df.show()

请求是找到最高薪酬员工的职位名称。输出应包括最高薪酬的职位名称或薪酬相同的多个职位。

输入：

worker:
|worker_id|first_name|last_name|salary|joining_date| department|
| 1| John| Doe| 10000| 2023–01–01|Engineering|
| 2| Jane| Smith| 12000| 2022–12–01| Marketing|
| 3| Alice| Johnson| 12000| 2022–11–01|Engineering|

title:
|worker_ref_id|worker_title|affected_from|
| 1| Engineer| 2022–01–01|
| 2| Manager| 2022–01–01|
| 3| Engineer| 2022–01–01|

输出：
|worker_id|first_name|last_name|best_paid_title|salary|
| 3| Alice| Johnson| Engineer| 12000|
| 2| Jane| Smith| Manager| 12000|

    从 pyspark.sql 导入 SparkSession, SQLContext, Window 函数  
    从 pyspark.sql.functions 导入 rank  

    spark = SparkSession.builder \  
        .appName("HighestPaidJobTitles") \  
        .getOrCreate()  

    worker_data = [(1, 'John', 'Doe', 10000, '2023-01-01', 'Engineering'),  
            (2, 'Jane', 'Smith', 12000, '2022-12-01', 'Marketing'),  
            (3, 'Alice', 'Johnson', 12000, '2022-11-01', 'Engineering')]  
    columns = ['worker_id', 'first_name', 'last_name', 'salary', 'joining_date', 'department']  
    worker = spark.createDataFrame(worker_data, columns)  

    title_data = [(1, 'Engineer', '2022-01-01'),  
            (2, 'Manager', '2022-01-01'),  
            (3, 'Engineer', '2022-01-01')]  
    columns = ['worker_ref_id', 'worker_title', 'affected_from']  
    title = spark.createDataFrame(title_data, columns)  

    joined_df = worker.join(title, worker.worker_id == title.worker_ref_id)  

    ranked_df = joined_df.withColumn("salary_rank", f.rank().over(Window.orderBy(joined_df["salary"].desc())))  
    highest_paid_df = ranked_df.filter(ranked_df["salary_rank"] == 1)  
    result_df = highest_paid_df.select("worker_id", "first_name", "last_name", "worker_title", "salary").withColumnRenamed('worker_title', 'best_paid_title')  
    result_df.show()  

    spark.stop()

您需要从以下示例数据中找出工资最高和最低的员工。输出包括一个salary_type列，该列将输出分类为：
‘最高工资员工’表示工资最高的员工；‘最低工资员工’表示工资最低的员工。

员工:
|worker_id|first_name|last_name|salary|joining_date| department|
| 1| John| Doe| 5000| 2023年01月01日|Engineering|
| 2| Jane| Smith| 6000| 2023年01月15日| Marketing|
| 3| Alice| Johnson| 4500| 2023年02月05日|Engineering|

职位:
|worker_ref_id|worker_title|affected_from|
| 1| Engineer| 2022年01月01日|
| 2| Manager| 2022年01月01日|
| 3| Engineer| 2022年01月01日|

    from pyspark.sql import SparkSession  
    from pyspark.sql.functions import max, min, when  

    spark = SparkSession.builder \  
        .appName("最高和最低薪水的员工") \  
        .getOrCreate()  

    worker_data = [  
        (1, 'John', 'Doe', 5000, '2023-01-01', '工程'),  
        (2, 'Jane', 'Smith', 6000, '2023-01-15', '市场营销'),  
        (3, 'Alice', 'Johnson', 4500, '2023-02-05', '工程')  
    ]  
    title_data = [  
        (1, '工程师', '2022-01-01'),  
        (2, '经理', '2022-01-01'),  
        (3, '工程师', '2022-01-01')  
    ]  
    worker_columns = ['worker_id', 'first_name', 'last_name', 'salary', 'joining_date', 'department']  
    title_columns = ['worker_ref_id', 'worker_title', 'affected_from']  
    worker_df = spark.createDataFrame(worker_data, worker_columns)  
    title_df = spark.createDataFrame(title_data, title_columns)  
    worker_df.show()  
    title_df.show()  

    joined_df = worker_df.join(title_df, worker_df.worker_id == title_df.worker_ref_id, "inner")  
    result_df = joined_df.groupBy("worker_id", "first_name", "last_name", "salary", "department") \  
        .agg(  
            max("salary").alias("max_salary"),  
            min("salary").alias("min_salary")  
        )  

    result_df = result_df.withColumn("salary_type",  
                    when(result_df["salary"] == result_df["max_salary"], "最高薪资")  
                    .when(result_df["salary"] == result_df["min_salary"], "最低薪资")  
                    .otherwise(None))  
    result_df.show()  

    spark.stop()

给定的输入 -
StudentID, StudentName , AScore, BScore, CScore
123, A, 30, 31, 32
124, B, 40, 41, 42

请按照以下格式输出 -
StudentID, StudentName , 科目 , Score
123, A, AScore, 30
123, A, BScore, 31
123, A, CScore, 32
124, B, AScore, 40
124, B, BScore, 41
124, B, CScore, 42

from pyspark.sql import SparkSession  
from pyspark.sql.functions import *  

spark = SparkSession.builder \  
    .appName("Transform Data") \  
    .getOrCreate()  

data = [  
    (123, "A", 30, 31, 32),  
    (124, "B", 40, 41, 42),  
    (125, "B", 50, 51, 52)  
]  

df = spark.createDataFrame(data, ["StudentID", "StudentName", "AScore", "BScore", "CScore"])  

pivot_df = df.selectExpr(  
    "StudentID",  
    "StudentName",  
    "stack(3, 'AScore', AScore, 'BScore', BScore, 'CScore', CScore) as (Subject, Score)"  
)  

pivot_df.show() # 显示转换后的数据框

注意：
"stack(3, 'AScore', AScore, 'BScore', BScore, 'CScore', CScore)"：这部分使用了 stack 函数将 DataFrame 转换为堆叠格式。第一个参数 3 表示要堆叠的层次数。后续的参数是键值对，其中第一个参数是键（本例中为列名），第二个参数是值（列的值）。因此，对于每一行中的数据，都会创建三个键值对，例如，对于某一行，将生成这样的键值对：('AScore', AScore)、('BScore', BScore) 和 ('CScore', CScore)。堆叠操作将这些键值对堆叠成新的行。

给定输入 —
ID
1
2
3

输出 —
ID
1
2
2
3
3
3

    从pyspark.sql导入SparkSession  
    从pyspark.sql.functions导入expr, explode, split  
    spark = SparkSession.builder \  
        .appName("重复ID") \  
        .getOrCreate()  
    df = spark.createDataFrame([(1,), (2,), (3,)], ["id"])  

    output_df = df.selectExpr("explode(序列(1, id)) as id")  

    output_df.show()

输入 —
col1|col2|col3
alpha| aa| 1
alpha| aa| 2
beta| bb| 3
beta| bb| 4
beta| bb| 5

输出 —
col1|col2|col3列表
alpha| aa| [1, 2]
beta| bb| [3, 4, 5]

    从pyspark.sql导入SparkSession as SparkSession  
    从pyspark.sql.functions导入*  

    spark = SparkSession.builder \  
        .appName("HighestLowestSalaryEmployees") \  
        .getOrCreate()  

    data = [("alpha", "aa", 1),  
            ("alpha", "aa", 2),  
            ("beta", "bb", 3),  
            ("beta", "bb", 5),  
            ("beta", "bb", 4)]  
    schema = ["col1", "col2", "col3"]  
    df = spark.createDataFrame(data, schema=schema)  
    df.show()  # 显示DataFrame内容  
    df_grouped = df.groupBy("col1", "col2").agg(collect_list("col3").alias("col3_list"))  
    df_grouped.show()  # 显示分组后的DataFrame内容  
    spark.stop()  # 停止Spark会话  

读取下面的 JSON 文件 —
[
{
“dept_id”: 102,
“e_id”: [
10201,
10202
]
},
{
“dept_id”: 101,
“e_id”: [
10101,
10102,
10103
]
}
]

输出结果 —
部门ID | 员工ID
101 | 10101
101 | 10102
101 | 10103
102 | 10201
102 | 10202

    from pyspark.sql import SparkSession  

    spark = SparkSession.builder \  
        .appName("ReadJSON") \  
        .getOrCreate()  

    df = spark.read.option("multiline", "true").json('sample_data.json')  
    df_exploded = df.selectExpr("dept_id", "explode(e_id) as e_id")  
    # df_exploded = df.select("dept_id", explode('e_id').alias('e_id'))  
    df_exploded.show()

注意：
必须使用 option("multiline", "true")，否则将引发以下异常 —
_AnalysisException: 从 Spark 2.3 开始，当引用的列仅包含内部的损坏记录列（默认命名为 _corruptrecord 的）时，从原始 JSON/CSV 文件中发起的查询将不允许。

这是因为PySpark认为每个JSON文件中的记录都被视为单行内的完整记录。PySpark的JSON数据源接口提供了multiline选项，可以用来从多行中读取记录。

根据以下数据，计算员工在办公室内的总小时数。

输入 —
emp_id| punch_time| 标记（flag）
11114|1900–01–01 08:30:00| I
11114|1900–01–01 10:30:00| O
11114|1900–01–01 11:30:00| I
11114|1900–01–01 15:30:00| O
11115|1900–01–01 09:30:00| I
11115|1900–01–01 17:30:00| O

    import datetime  
    from pyspark.sql.types import StructType, StructField, TimestampType, LongType, StringType  
    from pyspark.sql import SparkSession  
    from pyspark.sql.window import Window  
    from pyspark.sql.functions import *  

    spark = SparkSession.builder.appName("TotalInTime").getOrCreate()  

    _data = [  
        (11114, datetime.datetime.strptime('08:30:00.00', "%H:%M:%S.%f"), "I"),  
        (11114, datetime.datetime.strptime('10:30:00.00', "%H:%M:%S.%f"), 'O'),  
        (11114, datetime.datetime.strptime('11:30:00.00', "%H:%M:%S.%f"), 'I'),  
        (11114, datetime.datetime.strptime('15:30:00.00', "%H:%M:%S.%f"), 'O'),  
        (11115, datetime.datetime.strptime('09:30:00.00', "%H:%M:%S.%f"), 'I'),  
        (11115, datetime.datetime.strptime('17:30:00.00', "%H:%M:%S.%f"), 'O')  
    ]  

    # Schema  
    _schema = StructType([  
        StructField('emp_id', LongType(), True),  
        StructField('punch_time', TimestampType(), True),  
        StructField('flag', StringType(), True)  
    ])  

    df = spark.createDataFrame(data=_data, schema=_schema)  

    window_agg = Window.partitionBy('emp_id').orderBy(col('punch_time'))  

    df = df.withColumn('prev_time', lag(col('punch_time')).over(window_agg))  

    df = df.withColumn('time_diff', (col('punch_time').cast('long') - col('prev_time').cast('long'))/3600)  

    df = df.groupBy('emp_id').agg(sum(when(col('flag') == 'O', col('time_diff')).otherwise(0)).alias('total_time'))  
    df.show()

从给定的数据集中，提取经理及其下属员工的信息

输入示例 —
employee_id|first_name|manager_id
4529| Nancy| 4125
4238| John| 4329
4329| Martina| 4125
4009| Klaus| 4329
4125| Mafalda| NULL
4500| Jakub| 4529
4118| Moira| 4952
4012| Jon| 4952
4952| Sandra| 4529
4444| Seamus| 4329

输出示例 —
manager_id|manager_name|人数
4125| Mafalda| 2
4952| Sandra| 2
4329| Martina| 3
4529| Nancy| 2

    from pyspark.sql import SparkSession  
    from pyspark.sql.functions import *  

    spark = SparkSession.builder \  
        .appName("员工及其经理") \  
        .getOrCreate()  

    data = [('4529', '南希', '4125'),  
    ('4238','约翰', '4329'),  
    ('4329', '马蒂娜', '4125'),  
    ('4009', '克劳斯', '4329'),  
    ('4125', '马法尔达', 'NULL'),  
    ('4500', '雅库布', '4529'),  
    ('4118', '莫拉', '4952'),  
    ('4012', '乔恩', '4952'),  
    ('4952', '桑德拉', '4529'),  
    ('4444', '西穆斯', '4329')]  
    schema = ['employee_id', 'first_name', 'manager_id']  

    df = spark.createDataFrame(data=data, schema=schema)  

    # 自连接DataFrame以获取经理的名字  
    result_df = df.alias("e").join(df.alias("m"), col("e.manager_id") == col("m.employee_id"), "inner") \  
        .select(col("e.employee_id"), col("e.first_name"),  
                col("e.manager_id"), col("m.first_name").alias("manager_name"))  

    # 按经理ID和经理名字分组并计算每个组的员工数量  
    result_df.groupBy("manager_id", "manager_name").count().show()  

    spark.stop()

编写一个使用pyspark的单词计数函数，涵盖以下步骤 —
1. 从input.txt文件中读取数据，并将其转换为RDD
2. 将单词转换为小写
3. 移除标点符号
4. 移除None值、空字符串和数字
5. 按单词出现次数降序排序
6. 将排序后的单词及其计数写入csv文件

测试数据如下：
test_data = [
“Hello world! This is a test.”,
“Spark is awesome, isn’t it?”,
“Test test test.”,
“Test 0 21.”
]

输出 —
单词|出现次数
test| 4
is| 2
a| 1
awesome| 1
spark| 1
this| 1
world| 1
hello| 1
isn| 1
it| 1

    from pyspark.sql import SparkSession  
    from pyspark.sql.functions import *  
    from pyspark.sql.types import StringType  

    def word_count(output_file_path):  
        # 初始化 SparkSession  
        spark = SparkSession.builder \  
            .appName("WordCount") \  
            .getOrCreate()  

        # 测试数据  
        test_data = [  
            "Hello world! This is a test.",  
            "Spark is awesome, isn't it?",  
            "Test test test.",  
            "Test 0 21."  
        ]  

        # 从测试数据创建 DataFrame  
        input_df = spark.createDataFrame(test_data, StringType())  

        input_df = input_df.select(split(col("value"), " ").alias("line"))  
        input_df = input_df.select(explode(col("line")).alias("value"))  

        # 将单词转换为小写形式  
        input_df = input_df.withColumn("value", lower(col("value")))  

        # 将文本分割成单词，并移除标点符号  
        words_df = input_df.select(regexp_extract(col("value"), "[a-z]+", 0).alias("word"))  

        # 移除 None 或空字符串  
        words_df = words_df.filter(col("word").isNotNull() & (col("word") != "") & (~col("value").rlike("^\d+$")))  

        # 执行单词计数  
        word_count_df = words_df.groupBy("word").count()  

        # 按单词计数降序排列  
        word_count_df = word_count_df.orderBy(col("count").desc())  
        word_count_df.show()  

        # 将单词计数结果写入 CSV 文件  
        word_count_df.coalesce(1).write.csv(output_file_path, header=True)  

        # 停止 SparkSession  
        spark.stop()  

    # 使用示例:  
    output_file_path = "output_word_count_result.csv"  
    word_count(output_file_path)

我希望这些问题能让你了解所问问题的类型，并且有信心在 PySpark 面试中展示自己，展示你在处理大数据任务方面的专长。因为这些问题涵盖了各种不同的 PySpark 函数，这些函数真的很重要，你需要知道并理解它们的用法，以便能够生成所需的输出。是的，回答这些问题的方法不止一种，我仅分享了我的解答方式。

我将在接下来的PySpark面试问题系列文章中提出更多问题。希望这能鼓励你更深入地了解PySpark的各项功能，并进行动手实践。