pyspark中map函数内部的操作

Question

我想从文件名中获取数据（因为它包含一些信息。）并将这些数据写入 csvfile_info 文件而不使用 loop 。 我是 pyspark 的新手。请有人帮助我编写代码，让我知道我该如何继续。 这是我尝试过的......

代码： c = os.path.join("-------")

input_file = sc.textFile(fileDir)
file1= input_file.split('_')
csvfile_info= open(c,'a')
details= file1.map(lambda p:
    name=p[0], 
    id=p[1],
    from_date=p[2],
    to_date=p[3],
    TimestampWithExtension=p[4]\
    file_timestamp=TimestampWithExtension.split('.')[0]\
    info = '{0},{1},{2},{3},{4},{5} \n'.\
    format(name,id,from_date,to_date,file_timestamp,input_file)\
    csvfile_info.write(info)
    )

Answer 1

不要尝试在map() 函数中写入数据。相反，您应该将每条记录映射到适当的字符串，然后将生成的 rdd 转储到文件中。试试这个：

input_file = sc.textFile(fileDir)  # returns an RDD

def map_record_to_string(x):
    p = x.split('_')
    name=p[0]
    id=p[1]
    from_date=p[2]
    to_date=p[3]
    TimestampWithExtension=p[4]

    file_timestamp=TimestampWithExtension.split('.')[0]
    info = '{0},{1},{2},{3},{4},{5} \n'.format(
        name,
        id,
        from_date,
        to_date,
        file_timestamp,
        input_file
    )
    return info

details = input_file.map(map_record_to_string)  # returns a different RDD
details.saveAsTextFile("path/to/output")

注意：我尚未测试此代码，但这是您可以采用的一种方法。

---

说明

从docs，input_file = sc.textFile(fileDir) 将返回带有文件内容的字符串RDD。

您要做的所有操作都是针对 RDD 的内容，即文件的元素。在 RDD 上调用 split() 没有意义，因为 split() 是一个字符串函数。您想要做的是调用split() 以及对RDD 的每条记录（文件中的行）的其他操作。这正是map() 所做的。

RDD 就像一个可迭代对象，但您不能使用传统循环对其进行操作。这是一个允许并行化的抽象。从用户的角度来看，map(f) 函数将函数 f 应用于 RDD 中的每个元素，就像在循环中完成一样。函数调用input_file.map(f) 等价于：

# let rdd_as_list be a list of strings containing the contents of the file
map_output = []
for record in rdd_as_list:
    map_output.append(f(record))

或等价：

# let rdd_as_list be a list of strings containing the contents of the file
map_output = map(f, rdd_as_list)

在RDD上调用map()会返回一个新的RDD，其内容是应用函数的结果。在这种情况下，details 是一个新的 RDD，它包含 input_file 的行，这些行已经被 map_record_to_string 处理过。

如果这样更容易理解，您也可以将map() 步骤写为details = input_file.map(lambda x: map_record_to_string(x))。