Python-大数据

Python 在大数据方面的实践。如：

• Apache Spark：大规模数据处理的开源集群式通用并行运算框架，基于 map reduce 算法的分布式计算
• PySpark：Spark 的 Python 接口

概述

Apache Spark：大规模数据处理的开源集群式通用并行运算框架，基于 map reduce 算法的分布式计算

• 特点：

  • 相比 Hadoop MapReduce，Hadoop 有的，Spark 都有
  • 将中间输出于结果保存于内存，不需读写 DHFS，适用于机器学习与数据挖掘的多次迭代；MapReduce 需将中间数据写入磁盘
  • 分布式计算：分发数据 + 分发计算
  • HDFS 分布式数据
  • 基于内存，适用于反复操作次数多的。

• RDD 抽象：Spark 提供了一个接口，用于编程具有隐式数据并行和容错功能的集群。

• 内存运算：

• 通用集群计算
  它比其它的大数据框架的优势在于：
  1、速度：比传统的大型数据处理框架快 100 倍。
  2、强大的缓存：简单的编程层提供了强大的缓存和磁盘持久性功能。
  3、部署：可以通过 Mesos，通过 Yarn 的 Hadoop 或 Spark 自己的集群管理器进行部署。
  4、实时：由于内存中的计算，实时计算和低延迟。
  5、多语言：这是该框架最重要的特性之一，因为它可以在 Scala，Java，Python 和 R 语言中编程。

• 弹性分布式数据集 RDDs（resilient distributed datasets）的分布式数据框架
  Hadoop 的 MapReduce 或 YARNf 将数据从磁盘中加载出来，DAG 后再写入磁盘，中间过程要对 HDFS 临时文件进行读写

核心组件

• Spark Core：包含 Spark 的基本功能；尤其是定义 RDD 的 API、操作以及这两者上的动作。其他 Spark 的库都是构建在 RDD 和 Spark Core 之上的。
• Spark SQL：提供通过 Apache Hive 的 SQL 变体 Hive 查询语言（HiveQL）与 Spark 进行交互的 API。每个数据库表被当做一个 RDD，Spark SQL 查询被转换为 Spark 操作。对熟悉 Hive 和 HiveQL 的人，Spar k 可以拿来就用。
• Spark Streaming：允许对实时数据流进行处理和控制。很多实时数据库（如 Apache Store）可以处理实时数据。Spark Streaming 允许程序能够像普通 RDD 一样处理实时数据。
• MLlib：一个常用机器学习算法库，算法被实现为对 RDD 的 Spark 操作。这个库包含可扩展的学习算法，比如分类、回归等需要对大量数据集进行迭代的操作。之前可选的大数据机器学习库 Mahout，将会转到 Spark，并在未来实现。
• GraphX：控制图、并行图操作和计算的一组算法和工具的集合。GraphX 扩展了 RDD API，包含控制图、创建子图、访问路径上所有顶点的操作。

体系架构

Spark 体系架构包括如下三个主要组件：

• 数据存储：HDFS 文件系统，可存任何兼容于 Hadoop 的数据源，包括 DFS，Hbase，Cassandra 等。
• API： Scala，Java 和 Python 的标准 API 接口
• 管理框架：

名词解释

Application：用户自己写的 Spark 应用程序，批处理作业的集合。Application 的 main 方法为应用程序的入口，用户通过 Spark 的 API，定义了 RDD 和对 RDD 的操作。

SparkContext： Spark 最重要的 API，用户逻辑与 Spark 集群主要的交互接口，它会和 Cluster Master 交互，包括向它申请计算资源等。

Driver、Executeer：park 在执行每个 Application 的过程中会启动 Driver 和 Executor 两种 JVM 进程。Driver 进程为主控进程，负责执行用户 Application 中的 main 方法，提交 Job，并将 Job 转化为 Task，在各个 Executor 进程间协调 Task 的调度。运行在 Worker 上 的 Executor 进程负责执行 Task，并将结果返回给 Driver，同时为需要缓存的 RDD 提供存储功能。

资源：一个计算机节点为一个独立的计算资源。

安装

jdk 自带 openjdk

java -version
openjdk version "1.8.0_192"
OpenJDK Runtime Environment (build 1.8.0_192-b26)
OpenJDK 64-Bit Server VM (build 25.192-b26, mixed mode)

下载 spark2.4.0
配置环境

sudo vi /etc/profile
SPARK_HOME=/opt/spark-1.6.1-bin-hadoop2.6
PATH=${SPARK_HOME}/bin:$PATH
source /etc/profile
pyspark

PySpark

PySpark：Spark 的 Python 接口

pip install pyspark

from pyspark import SparkContext
import re
import sys
sc = SparkContext("local", "Max Temperature")
sc.textFile(
    sys.argv[1]).map(
    lambda s: s.split("\t")).filter(
    lambda rec: (rec[1] != "9999" and re.match("[01459]", rec[2]))).map(
    lambda rec: (int(rec[0]),
                 int(rec[1]))).reduceByKey(max).saveAsTextFile(
    sys.argv[2])

from pyspark import SparkContext

sc = SparkContext('local')
# spark 的启动类型，分为 local、Mesos、YARN、Standalone 模式
#第二个参数 2 代表的是分区数，默认为 1
old=sc.parallelize([1,2,3,4,5],2)
newMap = old.map(lambda x:(x,x**2))
newReduce = old.reduce(lambda a,b : a+b)
print(newMap.glom().collect())
print(newReduce)

from pyspark import SparkContext
sc = SparkContext('local', 'test')
logFile = "readme.md"
# logData = sc.textFile(logFile, 2).cache()
# numAs = logData.filter(lambda line: 'a' in line).count()
# numBs = logData.filter(lambda line: 'b' in line).count()
# print('Lines with a: %s, Lines with b: %s' % (numAs, numBs))

# 创建 SparkContext 对象：执行上下文

fileRDD = sc.textFile('hdfs://localhost:9000/test.txt')


def contains(line):
    return 'hello world' in line


# 外源数据读入，创建 fileRDD
filterRDD = fileRDD.filter(contains)
# 构建 fileRDD 和 filterRDD 的依赖关系，形成 DAG 图
filterRDD.cache()
# 计算
filterRDD.count()