1.RDD编程指南

1.概述

两个抽象：RDD和共享变量

1. RDD
   RDD是跨集群节点分区的元素集合.

• 可以并行操作
• 可以持久化到内存中
• 会自动从节点故障中恢复

2. 共享变量
   广播变量，可用于在所有节点的内存中缓存值，
   累加器，它们是仅“添加”到的变量，例如计数器和总和。

2.RDD

2.1.创建

创建 RDD 有两种方法：并行化 驱动程序中的现有集合，或引用外部存储系统中的数据集。

1. 并行集合

List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
JavaRDD<Integer> distData = sc.parallelize(data, 2);

2. 外部数据集

JavaRDD<String> distFile = sc.textFile("data.txt");

2.2.操作

分为两类：转换（从现有数据集创建新数据集）和行动（在对数据集运行计算后将值返回给驱动程序）。
Spark 中的所有转换都是惰性的，
默认情况下，每个转换后的 RDD 可能会在您每次对其运行操作时重新计算。但是，可以使用persist(或者 cache) 方法将 RDD持久化。

1. 简单示例

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
int totalLength = lineLengths.reduce((a, b) -> a   b);
lineLengths.persist(StorageLevel.MEMORY_ONLY());

• 第一行定义了来自外部文件的基本 RDD。此数据集未加载到内存中或以其他方式执行：lines仅是指向文件的指针。
• 第二行定义lineLengths为map转换的结果。同样，由于懒惰，不会立即lineLengths 计算。
• 最后，我们运行reduce，这是一个动作。
• 这将导致lineLengths在第一次计算后保存在内存中。

2. 将函数传递给 Spark
   Spark 的 API 在很大程度上依赖于在驱动程序中传递函数来在集群上运行。
   有两种方法可以创建此类函数：

• 在您自己的类中实现 Function 接口，可以作为匿名内部类或命名的内部类，并将其实例传递给 Spark。
• 使用lambda 表达式 来简洁地定义一个实现。

3. 了解闭包
   关于 Spark 的难点之一是在跨集群执行代码时了解变量和方法的范围和生命周期。
   修改其范围之外的变量的 RDD 操作可能是一个常见的混淆源。

int counter = 0;
JavaRDD<Integer> rdd = sc.parallelize(data);

// Wrong: Don't do this!!
rdd.foreach(x -> counter  = x);

println("Counter value: "   counter);

4. 使用键值对
   虽然大多数 Spark 操作适用于包含任何类型对象的 RDD，但少数特殊操作仅适用于键值对的 RDD。最常见的是分布式“shuffle”操作，例如通过键对元素进行分组或聚合。

JavaRDD<String> lines = sc.textFile("data.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(s -> new Tuple2(s, 1));
JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a   b);

5. 常见转换

6. 常见行动

7. 混洗
   是 Spark 用于重新分配数据的机制，以便跨分区以不同方式分组。这通常涉及跨执行器和机器复制数据，使洗牌成为一项复杂且成本高昂的操作。
   可能导致洗牌的操作包括重新分区操作，如 repartition 和 coalesce，ByKey操作（计数除外），如groupByKey 和 reduceByKey，以及 连接操作，如cogroup 和 join。
   Shuffle是一项昂贵的操作，因为它涉及磁盘 I/O、数据序列化和网络 I/O 。
8. RDD 持久性
   Spark 中最重要的功能之一是跨操作将数据集持久化（或缓存）在内存中。缓存是迭代算法和快速交互使用的关键工具。
   每个持久化的 RDD 可以使用不同的存储级别进行存储：

3.共享变量

两种有限类型的共享变量：广播变量和累加器。

3.1.广播变量

Broadcast<int[]> broadcastVar = sc.broadcast(new int[] {1, 2, 3});
broadcastVar.value();

3.2.累加器

LongAccumulator accum = jsc.sc().longAccumulator();
sc.parallelize(Arrays.asList(1, 2, 3, 4)).foreach(x -> accum.add(x));
accum.value();

可以通过继承AccumulatorV2来创建自己的类型。

这篇好文章是转载于：学新通技术网