云计算

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

本文主要是介绍PySpark：面试题（编码题）— 第一部分，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
面试场景的旅程

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

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

Q1. ClickStream（点击流）

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

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()
Q2. 最高薪资

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

输入：

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()
Q3. 最高和最低薪水是多少？

您需要从以下示例数据中找出工资最高和最低的员工。输出包括一个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()
Q4. 将列转置为行（数据透视）

给定的输入 -
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

Sol - 
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)。堆叠操作将这些键值对堆叠成新的行。

Q5. 重复的标识符

给定输入 —
ID
1
2
3

输出 —
ID
1
2
2
3
3
3

Sol - 
    从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()
Q6. 分组行

输入 —
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]

Sol - 
    从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 数据

读取下面的 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

Sol - 
    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选项，可以用来从多行中读取记录。

Q8. 员工总计工时（内部）

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

输入 —
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()
Q9. 有经理的员工？

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

输入示例 —
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()
Q10. 计算每个单词的出现次数

编写一个使用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的各项功能，并进行动手实践。

这篇关于PySpark：面试题（编码题）— 第一部分的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
栏目导航