准备数据

方法创建数据存储airlinesmall.csv数据集。这个12兆字节的数据集包含多家航空公司的29列航班信息，包括到达和起飞时间。在这个例子中，感兴趣的变量是ArrDelay(航班延误)，DepDelay(航班起飞延误)和距离(总飞行距离)。

ds = tabularTextDatastore(“airlinesmall.csv”，“TreatAsMissing”，“NA”）;ds。ReadSize = 1000; ds.SelectedVariableNames = {“ArrDelay”，“DepDelay”，“距离”}；

数据存储处理“NA”值，并将缺少的值替换为南默认值。的ReadSize属性允许您指定如何将数据划分为块。此外,SelectedVariableNames属性允许您仅处理感兴趣的指定变量，您可以使用这些变量进行验证预览．

预览(ds)

ans =8×3表ArrDelay DepDelay Distance ________ ________ ________ 8 12 308 8 1 296 21 20 480 13 12 296 4 -1 373 59 63 308 3 -2 447 11 -1 954

链式MapReduce调用

实现多迭代TSQR算法需要链连续mapreduce调用。为了演示一般的链接设计模式，本例使用了两个mapreduce迭代。map函数调用的输出被传递到一大组约简器中，然后这些约简器的输出成为下一个约简器的输入mapreduce迭代。

第一次MapReduce迭代

在第一次迭代中，map函数，tsqrMapper，接收一个数据块(第i个)，这是一个大小的表 $$N_{我}\times 3.$$．映射器计算 $$R$$矩阵，并将其作为中间结果存储。然后,mapreduce在将中间结果发送给reduce函数之前，通过唯一键聚合中间结果。因此,mapreduce发送所有中间体 $$R$$矩阵具有相同的键到相同的减速器。

由于减速机使用qr，这是一个内存中的MATLAB®函数，最好先确保 $$R$$矩阵适合内存。这个示例将数据集划分为八个分区。的mapreduce函数以块为单位读取数据，并将数据和一些元信息传递给map函数。的信息Input参数是map函数的第二个输入，它包含了生成键所必需的读偏移量和文件大小信息，

key = ceil(offset/fileSize/numPartitions)。

显示map函数文件。

函数tsqrMapper(data, info, intermKVStore) x = data{:，:};X (any(isnan(X)，2)，:) = [];删除缺失值[~， r] = qr(x,0);% intermKey = randi(4);%用于划分中间结果的随机整数键intermKey = computeKey(info, 8);添加(intermKVStore intermKey, r);函数key = computeKey(info, numPartitions) fileSize = info. fileSize;基础数据文件总大小的%partitionSize =文件大小/numPartitions;每个分区的%大小(以字节为单位)offset = info.Offset;当前读取字节的%偏移量key = ceil(offset/partitionSize);结束结束

reduce函数接收中间数据的列表 $$R$$矩阵，垂直连接它们，并计算 $$R$$连接矩阵的矩阵。

显示reduce函数文件。

函数tsqrReducer(intermKey, intermValIter, outKVStore) x = [];而(intermValIter.hasnext) x = [x;intermValIter.getnext];结束注意，这种方法假设连接的中间值合适%内存。考虑增加reduce任务的数量(增加% tsqrMapper中的分区数量)，如果可以，则添加更多迭代%不适合内存。[~， r] =qr(x,0);添加(outKVStore intermKey, r);结束

使用mapreduce要将map和reduce函数应用到数据存储，ds．

outds1 = mapreduce(ds， @tsqrMapper， @tsqrReducer);

******************************** * MAPREDUCE的进展  * ******************************** 地图地图10%减少0% 20% 0%减少0%减少0%地图30%减少0% 40%减少0%地图地图60%减少0% 70% 50%减少0%减少0%地图80%减少0% 90%减少0%地图地图100%减少11% 100% 100%减少0%减少22%地图100%减少33% 100%减少44%地图地图100%减少67% 100% 100%减少56%减少78%地图100%减少89% 100%减少100%

mapreduce返回一个输出数据存储，outds1，文件在当前文件夹。

第二次MapReduce迭代

第二次迭代使用第一次迭代的输出，outds1，作为输入。这个迭代使用了一个身份映射器，identityMapper，它只是使用一个键复制数据，“身份”．

显示标识映射文件。

函数identityMapper(data, info, intermKVStore)此映射器函数只是复制数据并将其添加到% intermKVStore作为中间值。x = data.Value{:，:};添加(intermKVStore“身份”, x);结束

在两个迭代中，减速器函数是相同的。使用map函数的单个键意味着mapreduce在第二次迭代中只调用reduce函数一次。

使用mapreduce将单位映射器和相同的减速器应用到第一个输出mapreduce调用。

outds2 = mapreduce(outds1， @identityMapper， @tsqrReducer);

******************************** * MAPREDUCE的进展  * ******************************** 地图地图11%减少0% 22% 0%减少0%减少0%地图33%减少0% 44%减少0%地图55%减少0% 66%减少0%地图地图88%减少0% 100% 77%减少0%减少0%地图100%减少100%

查看结果

从输出数据存储中读取最终结果。

R = readall(outds2);r.Value {:}

ans =3×3105× 0.1091 0.0893 0.5564 0 -0.0478 -0.4890 0 0 3.0130

本地函数

这里列出的是map和reduce函数mapreduce适用于数据。

函数tsqrMapper(data, info, intermKVStore) x = data{:，:};X (any(isnan(X)，2)，:) = [];删除缺失值[~， r] = qr(x,0);% intermKey = randi(4);%用于划分中间结果的随机整数键intermKey = computeKey(info, 8);添加(intermKVStore intermKey, r);函数key = computeKey(info, numPartitions) fileSize = info. fileSize;基础数据文件总大小的%partitionSize =文件大小/numPartitions;每个分区的%大小(以字节为单位)offset = info.Offset;当前读取字节的%偏移量key = ceil(offset/partitionSize);结束结束%-------------------------------------------------------------------------------函数tsqrReducer(intermKey, intermValIter, outKVStore) x = [];而(intermValIter.hasnext) x = [x;intermValIter.getnext];结束注意，这种方法假设连接的中间值合适%内存。考虑增加reduce任务的数量(增加% tsqrMapper中的分区数量)，如果可以，则添加更多迭代%不适合内存。[~， r] =qr(x,0);添加(outKVStore intermKey, r);结束%-------------------------------------------------------------------------------函数identityMapper(data, info, intermKVStore)此映射器函数只是复制数据并将其添加到% intermKVStore作为中间值。x = data.Value{:，:};添加(intermKVStore“身份”, x);结束%-------------------------------------------------------------------------------

参考