yield-bytes

  本文介绍Spark DataFrame、Spark SQL、Spark Streaming入门使用教程，这些内容将为后面几篇进阶的streaming实时计算的项目提供基本计算指引，本文绝大部分内容来自Spark官网文档(基于PySpark):
Spark DataFrame、Spark SQL、Spark Streaming

1、RDD、Spark DataFrame、Spark SQL、Spark Streaming

  RDD：大家最熟悉的数据结构，主要使用transmissions和actions 相关函数式算子完成数据处理和数理统计，例如map、reduceByKey，rdd没有定义 Schema（一般指未定义字段名及其数据类型）， 所以一般用列表索引号来指定每一个字段。 例如， 在电影数据的推荐例子中:

1

move_rdd.map(lambda line:line.split('|')).map(lambda a_list:(alist[0],a_list[1],a_list[2]))

  本文内容第一部分给出Pyspark常见算子的用法，第二部分则参考书籍《Python spark2.0 Hadoop机器学习与大数据实战》的电影推荐章节。本文内容为大数据实时分析项目提供基本的入门知识。

1、PySpark简介

  本节内容的图文一部分参考了这篇文章《PySpark 的背后原理 》，个人欣赏此博客作者，博文质量高，看完受益匪浅！Spark的内容不再累赘，可参考本博客《深入理解Spark》。PySpark的工作原理图示如下：

  在前面的文章，已经通过gevent实现高并发的协程，本文将详细讨论Python标准库异步IO——asyncio。在Python3.4中引入了协程的概念以及asyncio。asyncio底层调用yield from语法，将任务变成生成器后挂起，这种方式无法实现协程之间的自动切换，在Python3.5中正式确立引入了async和await 的语法，所有的这些工作都使得Python实现异步编程变得更容易上手。

1、 yield 实现协程

1.1 yield 同步执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import time
def consumer():
    send_msg=''
    while 1:
        # 3、consumer通过yield拿到producer发来的消息，又通过yield把结果send_msg返回给producer
        output=yield send_msg
        print('[consumer] consuming {}'.format(output))
        send_msg='ok'

def producer(consumer_obj,num):
    # 1、启动consumer()生成器
    next(consumer_obj)
    for i in range(1,num+1):
        print('[producer] producing {}'.format(i))
        
        # 2、通过send()切换到consumer()执行
        receive_msg=consumer_obj.send(i)
        print('[producer] received a message {}'.format(receive_msg))
    consumer_obj.close()
    
if __name__=='__main__':
    c=consumer()
    producer(c,5)

  在前面的博客《深入理解Spark》 深入探讨了Spark架构原理内容，该文提到Stage的划分，为什么要做Stage划分？是为了得到更小的Task计算单元，分发给Executor的线程运行，将规模庞大、多流程的计算任务划分为有序的小颗粒的计算单元，实现更高效的计算。那么Stage划分怎么实现？需依赖RDD(Resilient Distributed Datasets，弹性分布式数据集)，可以说，RDD是Spark最为核心的概念。本文内容部分参考了《弹性式数据集RDDs》以及《Spark官方文档》

  在前面博客文章里，已经把大数据实时分析项目在spark组件之前的各个组件原理、部署和测试都给出相关讨论，接下来是项目最核心的内容：实时计算部分，因为项目将使用spark streaming做微批计算（准实时计算），因此接下的文章内容将深入spark以及spark streaming架构原理，为后面实际计算编程做铺垫。

1、Spark 是什么？

  Spark是一种分布式的并行计算框架，什么是计算框架？所谓的计算（在数据层面理解）其实是用于数据处理和分析的一套解决方案，例如Python的Pandas，相信用过Pandas都很容易理解Pandas擅长做什么，加载数据、对数据进行各类加工、分析数据等，只不过Pandas只适合在单机上的、数据量百万到千万级的计算组件，而Spark则是分布式的、超大型多节点可并行处理数据的计算组件。

  Spark通常会跟MapReduce做对比，它与MapReduce 的最大不同之处在于Spark是基于内存的迭代式计算——Spark的Job处理的中间（排序和shuffling）输出结果可以保存在内存中，而不是在HDFS磁盘上反复IO浪费时间。除此之外，一个MapReduce 在计算过程中只有Map 和Reduce 两个阶段。而在Spark的计算模型中，它会根据rdd依赖关系预选设计出DAG计算图，把job分为n个计算阶段（Stage），因为它内存迭代式的，在处理完一个阶段以后，可以继续往下处理很多个阶段，而不只是两个阶段。
  Spark提供了超过80种不同的Transformation和Action算子，如map，reduce，filter，reduceByKey，groupByKey，sortByKey，foreach等，并且采用函数式编程风格，实现相同的功能需要的代码量极大缩小（尤其用Scala和Python写计算业务代码方面）。正是基于使用易用性，因此Spark能更好地用于基于分布式大数据的数据挖掘与机器学习等需要迭代的MapReduce的算法。Spark生态如下：

  在本博客前面的文章给出redis-cluster模式的配置和测试《一篇文章掌握redis-cluster原理及其部署、测试》，redis还有另外一种failover自动切换的部署方式，也即是本文给出的——Sentinel模式（哨兵模式），这两种方式部署的redis服务其实在普通的项目完全够用，例如个人在Django项目使用的Sentinel模式保证了”查询缓存服务以及一些频繁读取配置参数服务“的高可用。对于并发量大的需求，可以使用国内知名Codis——分布式Redis集群代理中间件，可配置规模更大的redis集群服务。

  python的元类使用场景一般在大型框架里面，例如Django的ORM框架、基于python实现的高级设计模式，元类的这部分内容相对晦涩，但也是作为python非常核心的知识点，通过解析其机制，有利于阅读和学习优秀中间件源代码的设计逻辑，在面向对象设计的重要性不言而喻。本博客后面的内容将会给出较为复杂的设计模式的文章，里面会出现较多的元类编程，因此有必要单独开一篇文章讨论python元类，相关内容将参考Stack Overflow上一篇很受欢迎的关于python metaclasses的文章：what-are-metaclasses-in-python

  在前面的《基于hadoop3.1.2分布式平台上部署spark HA集群》，这篇是基于非HA模式下hadoop集群的spark集群HA配置，而本文将给出基于HA模式下hadoop集群的spark集群HA配置，并将yarn HA集群映入到spark中，做资源管理。为何要做些环境的配置呢？因为到本篇文章为止，已经完成hadoop HA集群、hbaseHA集群，hive集群（非HA）、sparkHA集群、flumeHA集群、kafka HA集群，实现实时数据流动，接下的文章重点探讨spark streaming、spark以及pyspark相关知识，这将涉及多个计算任务以及相关计算资源的分配，因此需要借助yarn HA集群强大的资源管理服务来管理spark的计算任务，从而实现完整的、接近生产环境的、HA模式下的大数据实时分析项目的架构。

  在前面blog文章中：《在hadoopHA节点上部署flume高可用组件》和《在hadoopHA节点上部署kafka集群组件》，已经实现大数据实时数据流传输两大组件的部署和测试，本文将讨论flume组件连接kafka集群相关内容，两组件在项目架构图的位置如下图1红圈所示：

  flume NG集群前向的source是各类实时的log数据，通过flume sink将这些日志实时sink到后向kafka集群，所有flume sink其实是本架构里kafka的producer角色，kafka集群后向连接spark streaming，用于消费kafka的实时消息（log日志数据）流。