正文

大数据实习之spark

IP云
IP云V管理员/9 阅读
08/20
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

Apache Spark是一个围绕速度、易用性和复杂分析构建的大数据处理框架。

与 Hadoop 和 Storm 等其他大数据和 MapReduce 技术相比，Spark 有如下优势。

  • 首先，Spark 为我们提供了一个全面、统一的框架用于管理各种有着不同性质（文本数据、图表数据等）的数据集和数据源（批量数据或实时的流数据）的大数据处理的需求。
  • Spark 可以将 Hadoop 集群中的应用在内存中的运行速度提升 100 倍，甚至能够将应用在磁盘上的运行速度提升 10 倍。
  • Spark 让开发者可以快速的用 Java、Scala 或 Python 编写程序。它本身自带了一个超过 80 个高阶操作符集合。而且还可以用它在 shell 中以交互式地查询数据。
  • 除了 Map 和 Reduce 操作之外，它还支持 SQL 查询，流数据，机器学习和图表数据处理。开发者可以在一个数据管道用例中单独使用某一能力或者将这些能力结合在一起使用。

Spark 将中间结果保存在内存中而不是将其写入磁盘，当需要多次处理同一数据集时，这一点特别实用。

Spark 是用Scala 程序设计语言编写而成，运行于 Java 虚拟机（JVM）环境之上。目前支持如下程序设计语言编写 Spark 应用：

  • Scala
  • Java
  • Python
  • Clojure
  • R

Spark 生态系统

除了 Spark 核心 API 之外，Spark 生态系统中还包括其他附加库，可以在大数据分析和机器学习领域提供更多的能力。

这些库包括：

  • Spark Streaming:
    • Spark Streaming基于微批量方式的计算和处理，可以用于处理实时的流数据。它使用 DStream，简单来说就是一个弹性分布式数据集（RDD）系列，处理实时数据。
  • Spark SQL:
    • Spark SQL可以通过 JDBC API 将 Spark 数据集暴露出去，而且还可以用传统的 BI 和可视化工具在 Spark 数据上执行类似 SQL 的查询。用户还可以用 Spark SQL 对不同格式的数据（如 JSON，Parquet 以及数据库等）执行 ETL，将其转化，然后暴露给特定的查询。
  • Spark MLlib: 
    • MLlib是一个可扩展的 Spark 机器学习库，由通用的学习算法和工具组成，包括二元分类、线性回归、聚类、协同过滤、梯度下降以及底层优化原语。
  • Spark GraphX: 
    • GraphX是用于图计算和并行图计算的新的（alpha）Spark API。通过引入弹性分布式属性图（Resilient Distributed Property Graph），一种顶点和边都带有属性的有向多重图，扩展了 Spark RDD。为了支持图计算，GraphX 暴露了一个基础操作符集合（如 subgraph，joinVertices 和 aggregateMessages）和一个经过优化的 Pregel API 变体。此外，GraphX 还包括一个持续增长的用于简化图分析任务的图算法和构建器集合。

除了这些库以外，还有一些其他的库，如 BlinkDB 和 Tachyon。

BlinkDB是一个近似查询引擎，用于在海量数据上执行交互式 SQL 查询。BlinkDB 可以通过牺牲数据精度来提升查询响应时间。通过在数据样本上执行查询并展示包含有意义的错误线注解的结果，操作大数据集合。

Tachyon是一个以内存为中心的分布式文件系统，能够提供内存级别速度的跨集群框架（如 Spark 和 MapReduce）的可信文件共享。它将工作集文件缓存在内存中，从而避免到磁盘中加载需要经常读取的数据集。通过这一机制，不同的作业 / 查询和框架可以以内存级的速度访问缓存的文件。

函数式编程Lambda

Lambda的应用场景

1.使用() -> {} 替代匿名类

现在Runnable线程，Swing，JavaFX的事件监听器代码等，在java 8中你可以使用Lambda表达式替代丑陋的匿名类。

//Before Java 8:
new Thread(new Runnable() {
    @Override
    public void run() {
        System.out.println("Before Java8 ");
    }
}).start();
 
//Java 8 way:
new Thread(() -> System.out.println("In Java8!"));
 
// Before Java 8:
JButton show =  new JButton("Show");
show.addActionListener(new ActionListener() {
     @Override
     public void actionPerformed(ActionEvent e) {
           System.out.println("without lambda expression is boring");
        }
     });
 
// Java 8 way:
show.addActionListener((e) -> {
    System.out.println("Action !! Lambda expressions Rocks");
});

2.使用内循环替代外循环

外循环：描述怎么干，代码里嵌套2个以上的for循环的都比较难读懂；只能顺序处理List中的元素；

内循环：描述要干什么，而不是怎么干；不一定需要顺序处理List中的元素

//Prior Java 8 :
List features = Arrays.asList("Lambdas", "Default Method",
"Stream API", "Date and Time API");
for (String feature : features) {
   System.out.println(feature);
}
 
//In Java 8:
List features = Arrays.asList("Lambdas", "Default Method", "Stream API",
 "Date and Time API");
features.forEach(n -> System.out.println(n));
 
// Even better use Method reference feature of Java 8
// method reference is denoted by :: (double colon) operator
// looks similar to score resolution operator of C++
features.forEach(System.out::println);
 
Output:
Lambdas
Default Method
Stream API
Date and Time API

3.支持函数编程

为了支持函数编程，Java 8加入了一个新的包java.util.function，其中有一个接口java.util.function.Predicate是支持Lambda函数编程：

public static void main(args[]){
  List languages = Arrays.asList("Java", "Scala", "C++", "Haskell", "Lisp");
 
  System.out.println("Languages which starts with J :");
  filter(languages, (str)->str.startsWith("J"));
 
  System.out.println("Languages which ends with a ");
  filter(languages, (str)->str.endsWith("a"));
 
  System.out.println("Print all languages :");
  filter(languages, (str)->true);
 
   System.out.println("Print no language : ");
   filter(languages, (str)->false);
 
   System.out.println("Print language whose length greater than 4:");
   filter(languages, (str)->str.length() > 4);
}
 
 public static void filter(List names, Predicate condition) {
    names.stream().filter((name) -> (condition.test(name)))
        .forEach((name) -> {System.out.println(name + " ");
    });
 }
 
Output:
Languages which starts with J :
Java
Languages which ends with a
Java
Scala
Print all languages :
Java
Scala
C++
Haskell
Lisp
Print no language :
Print language whose length greater than 4:
Scala
Haskell

4.处理数据？用管道的方式更加简洁

Java 8里面新增的Stream API ，让集合中的数据处理起来更加方便，性能更高，可读性更好

假设一个业务场景：对于20元以上的商品，进行9折处理，最后得到这些商品的折后价格。

final BigDecimal totalOfDiscountedPrices = prices.stream()
.filter(price -> price.compareTo(BigDecimal.valueOf(20)) > 0)
.map(price -> price.multiply(BigDecimal.valueOf(0.9)))
.reduce(BigDecimal.ZERO,BigDecimal::add);
 
System.out.println("Total of discounted prices: " + totalOfDiscountedPrices);

想象一下：如果用面向对象处理这些数据，需要多少行？多少次循环？需要声明多少个中间变量？

Lambda测试：

import java.util.Arrays;
import java.util.List;import org.junit.Test;public class LambdaTest {@Test
public void testInnerClass() throws InterruptedException {
//Before Java 8:
new Thread(new Runnable() {
@Override
public void run() {
System.out.println("Before Java8 ");
}
}).start();
new Thread(() -> System.out.println("After Java8")).start();Thread.sleep(1000L);
}
}

List<String> features = Arrays.asList("Lambdas","123","456","789");
for(String feature : features) {
System.out.println(feature);
}