2021-2022-2-编程作业2（不全）¶

2019级，Spark

Monte Carlo¶

题目描述¶

蒙特卡罗（ Monte Carlo ）算法计算圆周率的主要思想如下：给定边长为R的正方形，画其内切圆，然后在正方形内随机打点，设点落在圆内的概为P，则根据概率学原理： P = 落在圆内点的数量/正方形内点的数量 = 圆面积 / 正方形面积＝ PI * R * R / 2R * 2R = PI / 4。即 PI=4P。这样，当随机打点足够多时，统计出来的概率就非常接近于PI的四分之一了。请根据蒙特卡洛思想来估计 Pi 的值。在 DSPPCode.spark.pi 中创建 PiSimulatorImpl, 继承 PiSimulator, 实现抽象方法

PiSimulatorImpl¶

package DSPPCode.spark.pi.impl;

import DSPPCode.spark.pi.question.PiSimulator;
import org.apache.spark.api.java.JavaRDD;
import java.util.Random;

public class PiSimulatorImpl extends PiSimulator {
  @Override
  public JavaRDD<Integer> sampledPoint(JavaRDD<Integer> parallelInput) {
    return parallelInput.map(Integer ->
        (Math.pow(Math.pow(new Random().nextFloat(), 2) + Math.pow(new Random().nextFloat(), 2), 0.5) <= 1) ? 1 : 0);
  }
}

Friendship¶

题目描述¶

根据输入文本中的用户关系，找出任意两个用户之间的共同好友。输入格式：输入文本中每行表示一个用户关系，一个用户关系由多个列组成。其中，第一列为用户名，其余列为该用户的好友的用户名，用户名之间用空格分隔。例如，下面第一行表示用户A有B、C、D、 E 、F五位好友。

A B C D E F
B A C D E
C A B E
D A B E
E A B C D
F A

输出格式: 输出文本的每行表示两个用户的用户名及其所有共同好友的用户名，其中两个用户的用户名按升序字典序进行拼接，所有共同好友的用户名之间以逗号和空格分隔。例如，下面第一行表示用户A和C的两个用户的所有共同好友为B、E。

(AC,[B, E])
(BC,[A, E])
(CE,[A, B])
(AB,[C, D, E])
(AE,[B, C, D])
(CD,[A, B, E])
(DF,[A])
(CF,[A])
(AD,[B, E])
(DE,[A, B])
(BE,[A, C, D])
(BF,[A])
(EF,[A])
(BD,[A, E])

在 DSPPCode.spark.friendship.impl 中创建 FriendshipImpl, 继承 Friendship, 实现抽象方法。注意：请先完成orderedPairs函数。

FriendshipImpl¶

package DSPPCode.spark.friendship.impl;

import DSPPCode.spark.friendship.question.Friendship;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import scala.Tuple2;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class FriendshipImpl extends Friendship {
  @Override
  public JavaPairRDD<String, Iterable<String>> findFriendShip(JavaRDD<String> lines) {
    JavaPairRDD<String, String> RelRdd = lines.flatMapToPair(
        new PairFlatMapFunction<String, String, String>() {
          @Override
          public Iterator<Tuple2<String, String>> call(String s) throws Exception {
            List<Tuple2<String, String>> resultList = new ArrayList<>();
            String[] RelList = s.split(" ");
            List<String> keyList = orderedPairs(Arrays.copyOfRange(RelList, 1, RelList.length));
            for(String key: keyList){
              resultList.add(Tuple2.apply(key, RelList[0]));
            }
            return resultList.iterator();
          }
        });
    return RelRdd.groupByKey();
  }

  @Override
  public List<String> orderedPairs(String[] strs) {
    List<String> res = new ArrayList<>();
    Arrays.sort(strs);
    for(int i = 0; i < strs.length; i++){
      for(int j = i + 1; j < strs.length; j++){
        res.add(strs[i] + strs[j]);
      }
    }
    return res;
  }
}

Transitive Closure¶

题目描述¶

在数学中，集合 X 上的二元关系 R 的传递闭包指的是包含 R 的 X 上的最小的传递关系，记作 t®。例如，假设集合 X 为人的集合 {a,b,c}，二元关系 R 为父子关系 {,}，其中 和 分别表示a是b的父亲以及b是c的父亲，则 t® 应为祖宗-后代关系 {,,}。当前，社保局拿到了一份名单，该名单给出了子女-父母的关系。社保局想要从该名单中分析出名单中包含的子女-祖父母、外祖父母关系。然而，名单很庞大，如果手工分析可能需要几个小时，于是社保局希望你能帮忙编写程序自动地进行分析。输入格式:

child parent
Jack Philip
Jack Jesse
Philip Terry
Philip Alma
Jesse Anna
Jesse John

输入保存在文本中，文本的第一行为关系标识，其余行为子女和父母的对应关系。以上述示例为例，Jack Philip表示Jack与Philip之间存在着子女-父母的关系。输出格式:

(Jack,[Anna, John, Terry, Alma])

输出保存在文本中，每行的第一个名字代表子女，后面跟随祖父母、外祖父母共四个名字，名字之间使用空格隔开。需要注意的是，由于输入数据可能缺失部分记录，对于无法分析出祖父母、外祖父母共四人的孙子女，社保局要求不展示在输出结果中。以上述示例为例，(Jack,[Anna, John, Terry, Alma])表示Jack与Terry、Alma、Anna、John之间存在着孙子女-祖父母（或外祖父母）关系。

在 DSPPCode.spark.transitive_closure.impl 中创建 TransitiveClosureImpl, 继承TransitiveClosuresImpl, 实现抽象方法

TransitiveClosureImpl¶

package DSPPCode.spark.transitive_closures.impl;

import DSPPCode.spark.transitive_closures.question.TransitiveClosures;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;


public class TransitiveClosuresImpl extends TransitiveClosures {

  @Override
  protected JavaPairRDD<String, Iterable<String>> transitive(JavaRDD<String> lines) {
    JavaPairRDD<String, String> childParentRdd = lines.mapToPair(
        new PairFunction<String, String, String>() {
          @Override
          public Tuple2<String, String> call(String s) throws Exception {
            String[] tokens = s.split(" ");
            String child = tokens[0];
            String parent = tokens[1];
            return new Tuple2<>(child, parent);
          }
        });

    JavaPairRDD<String, String> parentChildRdd = lines.mapToPair(
        new PairFunction<String, String, String>() {
          @Override
          public Tuple2<String, String> call(String s) throws Exception {
            String[] tokens = s.split(" ");
            String child = tokens[0];
            String parent = tokens[1];
            return new Tuple2<>(parent, child);
          }
        });

    JavaPairRDD<String, Tuple2<String, String>> temp1 = parentChildRdd.join(childParentRdd);
    JavaPairRDD<String, String> temp2 = temp1.values().mapToPair((Tuple2<String, String> t) -> new Tuple2<>(t._1, t._2));
    JavaPairRDD<String, Iterable<String>> res = temp2.groupByKey();
    return res.filter((Tuple2<String, Iterable<String>> t) -> (t._2).toString().split(",").length == 4);
  }
}