Python函数式编程与数据处理实践

1、哪个句子最能描述map函数？  map将一个数据序列转换为不同但大小相同的序列。  map允许我们有条件地处理数据，替代if – else语句。  map用优化的字节码替换条件while循环。

map将一个数据序列转换为不同但大小相同的序列。

2、并行计算很有用，因为它能让我们更快地处理大型数据集。以下哪一项解释了并行计算的工作原理？A. 并行计算在编译时优化我们的代码。B. 并行计算在多个计算资源上计算相似的任务。C. 并行计算从我们的代码中去除重复部分并减少昂贵操作的数量。

B

3、以下哪一项不是 pickle（序列化）的良好用途？A. 短期单机器存储；B. 在集群的计算任务之间共享数据；C. 对数据完整性要求高的长期存储

C

4、在网页抓取中，常见的操作之一是将字典转换为其他形式。使用

map

函数将以下列表中的字典转换为仅包含页面文本的列表。输入数据为：

[{'headers': ('01/19/2018', 'Mozilla', 300), 'response': {'text': 'Hello world!', 'encoding': 'utf - 8'}}, {'headers': ('01/19/2018', 'Chrome', 404), 'response': {'text': 'No page found', 'encoding': 'ascii'}}, {'headers': ('01/20/2018', 'Mozilla', 300), 'response': {'text': 'Yet another web page.', 'encoding': 'utf - 8'}}]

，最终结果列表应为

['Hello world!', 'No page found', 'Yet another web page.']


以下是使用 Python 实现该功能的代码：

input_data = [{'headers': ('01/19/2018', 'Mozilla', 300), 'response': {'text': 'Hello world!', 'encoding': 'utf - 8'}}, {'headers': ('01/19/2018', 'Chrome', 404), 'response': {'text': 'No page found', 'encoding': 'ascii'}}, {'headers': ('01/20/2018', 'Mozilla', 300), 'response': {'text': 'Yet another web page.', 'encoding': 'utf - 8'}}]

# 定义一个函数来提取文本
def extract_text(item):
    return item['response']['text']

result = list(map(extract_text, input_data))
print(result)

5、到目前为止，我们只研究了使用map来转换包含相同类型数据的同质列表。不过，我们没有理由不能使用map来转换异质数据列表。编写一个函数，将[1, “A”, False]转换为[2, “B”, True]。

可以使用Python编写如下函数：

def transform_list(lst):
    def transform(item):
        if isinstance(item, int):
            return item + 1
        elif isinstance(item, str):
            return chr(ord(item) + 1)
        elif isinstance(item, bool):
            return not item
        return item
    return list(map(transform, lst))

input_list = [1, "A", False]
output_list = transform_list(input_list)
print(output_list)

6、用你自己的话，定义辅助函数（helper functions）和函数管道（function pipelines）这两个术语，然后描述它们之间的关系。

辅助函数是用于完成复杂任务的小型、简单函数，可将大问题分解为易编码的小问题。函数管道（也叫函数链）是将辅助函数按顺序组合起来工作的方式。二者关系密切，函数管道依赖辅助函数来实现具体功能，通过将多个辅助函数连接起来，完成从输入到输出的复杂转换。

7、一个经典的数学老师的技巧是让学生对一个“未知”数字进行一系列算术运算，最后老师猜出学生所想的数字。诀窍在于最终的数字总是老师事先知道的一个常数。例如，将一个数字翻倍，加上10，再除以2，然后减去原来的数字。使用一系列链接在一起的小辅助函数，对1到100之间的所有数字进行这个过程。老师是如何总能知道你所想的数字的呢？

设这个未知数字为 $ x $，根据运算步骤可得：

((2x+10)÷2−x)((2x+10)÷2−x)

化简该式：

(2x+10)÷2−x=x+5−x=5(2x+10)÷2−x=x+5−x=5

所以无论学生想的数字是多少，经过这一系列运算后结果总是 5，老师事先知道这个常数是 5，就能猜出结果。

8、凯撒密码是一种古老的构建密码的方法，即将字母位置移动3位，比如A变成D，B变成E，C变成F，依此类推。将三个函数链接起来创建这个密码：一个将字母转换为整数，一个给数字加3，一个将数字转换为字母。通过对字符串应用链接后的函数，将此密码应用到一个单词上。创建一个新函数和新管道来解密你的密码。

要实现凯撒密码加密和解密，可按以下步骤进行。首先，定义三个关键函数：

一个用于将字母转换为整数

一个用于给数字加3

一个用于将数字转换为字母

然后将这三个函数链接起来，对输入的字符串进行加密。对于解密，创建一个新函数和新管道，将加3操作改为减3。

以下是 Python 代码示例：

# 将字母转换为整数
def letter_to_int(letter):
    return ord(letter.lower()) - ord('a')

# 给数字加3
def add_three(num):
    return (num + 3) % 26

# 将数字转换为字母
def int_to_letter(num):
    return chr(num + ord('a'))

# 加密函数
def caesars_cypher(word):
    return ''.join(int_to_letter(add_three(letter_to_int(letter))) for letter in word)

# 解密函数
def reverse_cypher(word):
    def subtract_three(num):
        return (num - 3) % 26
    return ''.join(int_to_letter(subtract_three(letter_to_int(letter))) for letter in word)

可以使用以下方式测试：

words = ['this', 'is','my','sentence']
encrypted = [caesars_cypher(word) for word in words]
print(encrypted)

# 解密
decrypted = [reverse_cypher(word) for word in encrypted]
print(decrypted)

以上代码定义了所需的函数，实现了凯撒密码的加密和解密功能。

9、在 Python 中使用映射和归约风格时，惰性函数很常见。以下哪些函数是惰性的？选项：map、reduce、filter、list、zip、sum、range、len

map、filter、zip、range

10、当我们使用像 range 这样的内置生成器函数时，只能对其进行一次迭代。为什么会这样呢？

当一个元素被生成后，Python 会将其遗忘，所以只能获取剩余的元素，因此只能对其进行一次迭代。

11、流行游戏拼字游戏（Scrabble）是通过在棋盘上放置字母牌来拼写单词。拼写长单词和包含更多稀有字母的单词能获得更多分数。在简化版中，字母 Z 价值 10 分；F、H、V 和 W 价值 5 分；B、C、M 和 P 价值 3 分；其他所有字母价值 1 分。使用 map 和 filter 函数，从以下单词列表中筛选出分数超过 8 分的单词：zebra, fever, charm, mouse, hair, brill, thorn。

计算单词分数并筛选

首先，我们需要定义一个函数来计算每个单词的分数，然后使用

map

函数将这个函数应用到单词列表上，得到每个单词的分数。接着，使用

filter

函数筛选出分数超过 8 分的单词。

以下是实现此功能的 Python 代码：

from functools import reduce

# 定义计算单词分数的函数
def score_word(word):
    points = 0
    for char in word:
        if char.lower() == 'z':
            points += 10
        elif char.lower() in 'fhvw':
            points += 5
        elif char.lower() in 'bcmp':
            points += 3
        else:
            points += 1
    return points

words = ['zebra', 'fever', 'charm', 'mouse', 'hair', 'brill', 'thorn']

# 使用 map 函数计算每个单词的分数
scores = map(score_word, words)

# 结合单词和分数
word_scores = zip(words, scores)

# 使用 filter 函数筛选出分数超过 8 分的单词
filtered_words = [word for word, score in word_scores if score > 8]

print(filtered_words)

运行这段代码，输出结果将是分数超过 8 分的单词列表：

['zebra', 'fever', 'charm', 'brill', 'thorn']

12、reduce函数是一个强大且灵活的工具。在以下哪些情况下你会使用reduce，哪些情况应该使用其他工具？1. 你有一长串单词，要返回仅包含字母A的序列。2. 你有一个用户序列，想把他们转换为仅包含用户ID号的序列。3. 你有一系列用户，想找出购买最多的五名用户。4. 你有一个采购订单序列，想找出采购的平均价格。

- 情况1和2可使用其他工具；情况3和4可使用reduce函数。
  - 情况1可使用过滤操作；
  - 情况2可使用映射操作；
  - 情况3需要对用户购买量排序并取前五，
  - 情况4需要对采购订单价格求和再求平均，这两种情况适合用reduce函数处理。

13、我们可以为只打算使用一次的简单函数使用 lambda 函数。请用 lambda 函数重写以下函数：def my_addition(a, b): return a+b；def is_odd(a): return a % 2 == 1；def contains(a, b): return b in a；def reverse(s): return s[::-1]。

下面是给定的【文本内容】：

my_addition

函数可重写为：

lambda a, b: a + b


is_odd

函数可重写为：

lambda a: a % 2 == 1


contains

函数可重写为：

lambda a, b: b in a


reverse

函数可重写为：

lambda s: s[::-1]


14、分组是一种有用的归约操作，我们可以对序列中的元素应用某个函数，然后根据函数的结果对元素进行分组。请使用归约（reduce）函数编写一个版本的函数，该函数可以根据单词的长度对单词进行分组。

以下是使用 Python 的

reduce

函数实现根据单词长度对单词进行分组的代码：

from functools import reduce

def group_words(words):
    def group_accumulator(acc, word):
        length = len(word)
        acc[length] = acc.get(length, []) + [word]
        return acc
    return reduce(group_accumulator, words, {})

你可以使用以下方式调用这个函数：

print(group_words(['these', 'are', 'some', 'words', 'for', 'grouping']))

运行上述代码，你将得到一个字典，其中键是单词的长度，值是具有该长度的单词列表。

15、我们为Hadoop Streaming作业编写的脚本叫什么？（单选）A. 映射器和归约器B. 应用器和累加器C. 函子和折叠器

A

16、当我们与Spark交互时，会通过PySpark进行，PySpark是围绕用哪种编程语言编写的Spark代码的Python包装器？（选择一项）A. ClojureB. ScalaC. JavaD. KotlinE. Groovy

B

17、Spark的创新围绕着一种叫做RDD的数据结构。RDD代表什么？（选择一项）A. 弹性分布式数据集B. 可靠定义的数据C. 可归约持久定义

A

18、使用 Hadoop Streaming 时，我们需要手动确保数据能够在映射和归约步骤之间传递。在每个步骤结束时我们需要调用什么？（选择一项）- 返回（return）- 生成（yield）- 打印（print）- 跳过（pass）

print

19、MRJob在映射步骤和归约步骤之间使用哪种数据格式共享数据？（选择一项）A. 二进制 B. 纯文本 C. JSON D. Pickle

C

20、判断题：像Hadoop MapReduce作业这样的并行进程是确定性的——它们的输出总是以相同的顺序产生。

错误。并行处理中通常存在多个任务同时执行，任务完成的顺序可能会受到多种因素影响，如系统负载、资源分配等，所以输出不一定总是以相同顺序产生。

21、以下哪种语言是 Hadoop 所使用的编程语言？（单选）A. Haskell B. C++ C. JavaScript D. Java

D

22、在使用 MRJob 时，如果我们尝试以 MRJob 风格进行编码——结合使用键和值以及映射器和归约器，能获得更好的性能。请使用 MRJob 实现以下常见的映射和归约风格模式：1. 过滤：获取一个序列并返回该序列的一个子集。示例：filter(is_even, [1,2,3,4,5]) 结果为 [2,4]；2. 频率统计：获取一个序列并统计该序列中元素的出现次数。示例：frequencies([1,2,1,1,2]) 结果为 {1:3, 2:2}；3. 分组：根据一个函数的结果对序列进行分组。示例：group_by(is_even, [1,2,3,4,5]) 结果为 {True: [2,4], False: [1,3,5]}；4. 按键计数：获取一个函数产生的键的计数。示例：count_by(is_even, [1,2,3,4,5]) 结果为 {True: 2, False: 3}。

以下是使用 MRJob 实现这些常见模式的示例代码：

from mrjob.job import MRJob
from functools import reduce

# 过滤模式
class MRFilter(MRJob):
    def mapper(self, _, line):
        num = int(line)
        if num % 2 == 0:
            yield num, None

    def reducer(self, key, values):
        yield key, None

# 频率统计模式
def make_counts(acc, nxt):
    acc[nxt] = acc.get(nxt, 0) + 1
    return acc

class MRFrequencies(MRJob):
    def mapper(self, _, line):
        num = int(line)
        yield num, 1

    def reducer(self, key, values):
        yield key, sum(values)

# 分组模式
class MRGroupBy(MRJob):
    def mapper(self, _, line):
        num = int(line)
        is_even = num % 2 == 0
        yield is_even, num

    def reducer(self, key, values):
        yield key, list(values)

# 按键计数模式
class MRCountBy(MRJob):
    def mapper(self, _, line):
        num = int(line)
        is_even = num % 2 == 0
        yield is_even, 1

    def reducer(self, key, values):
        yield key, sum(values)

if __name__ == '__main__':
    # 运行过滤作业
    MRFilter.run()
    # 运行频率统计作业
    MRFrequencies.run()
    # 运行分组作业
    MRGroupBy.run()
    # 运行按键计数作业
    MRCountBy.run()

你可以将输入数据保存为一个文本文件，每行一个数字，然后使用以下命令运行这些作业：

python your_script.py input.txt

其中

your_script.py

是上述代码保存的文件名，

input.txt

是包含输入数据的文件。

23、在Spark中，有一个由二元组组成的键值对RDD。需要实现一个操作，对这些键值对按键对所有值求和，将这个操作封装为一个名为sumByKey的函数。输入为一个包含多个键值对的RDD，调用sumByKey函数处理该RDD后，应输出按键求和后的结果。例如，若输入的RDD包含键值对 [(‘A’, 1), (‘A’, 1), (‘A’, 2), (‘B’, 2), (‘A’, 1), (‘C’, 1), (‘D’, 7), (‘D’, -2)]，处理后应得到 [(‘A’, 5), (‘B’, 2), (‘C’, 1), (‘D’, 5)]。

可以使用

reduceByKey

方法来实现

sumByKey

函数，示例代码如下：

from pyspark import SparkContext
sc = SparkContext()

def sumByKey(rdd):
    return rdd.reduceByKey(lambda x, y: x + y).collect()

xs = sc.parallelize([("A", 1), ("A", 1), ("A", 2), ("B", 2), ("A", 1), ("C", 1), ("D", 7), ("D", -2)])
result = sumByKey(xs)
print(result)

此代码定义了

sumByKey

函数，使用

reduceByKey

方法按键对值求和，最后使用

collect

方法将结果收集到驱动程序并打印。

24、对于我们知道很少会用到且有足够预警时间的数据，哪种S3存储类是最佳选择？

S3 Glacier

25、对于每月只需要使用几次的数据，哪种S3存储类别最合适？

S3 Infrequent Access

26、对于我们需要经常访问的数据，哪种S3存储类是最佳选择？

S3 Standard存储类

最适合需要经常访问的数据。它存储成本最高，但每次交易成本最低，适合频繁使用的数据。

27、AWS资源存在于可用区或区域中。高度耐用的S3存储是存在于可用区还是跨区域呢？

跨区域

28、对象存储的三个不可或缺的组成部分是哪三个？选项A：对象、对象名称、对象位置；选项B：对象、对象路径、对象颜色；选项C：对象、对象大小、元数据；选项D：对象、对象ID、元数据

D

29、编写一个mrjob配置文件，用于启动一个由五个R系列实例组成的集群。

# 以下是一个示例mrjob配置文件：
runners:
  emr:
    num_core_instances: 5
    instance_type: rX.large  # 这里的X根据具体R系列实例类型替换，例如r5
    region: us-west-1  # 可根据需要修改区域
    tags:
      project: Mastering Large Datasets

30、Spark的一大优点是它有很多便捷方法。在Scala中，复现以下使用toolz编写的操作。可使用Spark风格的方法链式调用以提高可读性。代码如下：

import toolz
xs = [('orange', 'O'), ('apple', 'A'), ('tomato', 'T'), ('kiwi', 'K'), ('lemon', 'L')]
toolz.take(10, toolz.frequencies(filter(lambda x: 'a' in x[0], xs)))

以下是使用Scala和Spark实现上述操作的代码示例：

import org.apache.spark.sql.SparkSession

object SparkToolzReplication {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("ToolzReplication")
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext
    val xs = List("orange" -> "O", "apple" -> "A", "tomato" -> "T", "kiwi" -> "K", "lemon" -> "L")
    val result = sc.parallelize(xs)
      .filter(_._1.contains("a"))
      .map(_._2)
      .countByValue()
      .toList
      .take(10)
    result.foreach(println)
    spark.stop()
  }
}

此代码首先创建了一个

SparkSession

，将列表

xs

转换为RDD，然后使用

filter

方法筛选出键中包含字母

'a'

的元素，接着使用

map

方法提取值，再用

countByValue

方法统计值的频率，最后使用

take

方法获取前10个结果。